You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by fa...@apache.org on 2013/04/08 17:19:09 UTC
svn commit: r1465662 [19/26] - in /qpid/trunk/qpid/tools/src/java: ./ bin/
bin/qpid-web/ bin/qpid-web/authentication/ bin/qpid-web/web/
bin/qpid-web/web/itablet/ bin/qpid-web/web/itablet/css/
bin/qpid-web/web/itablet/images/ bin/qpid-web/web/itablet/im...
Added: qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/ConnectionHelper.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/ConnectionHelper.java?rev=1465662&view=auto
==============================================================================
--- qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/ConnectionHelper.java (added)
+++ qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/ConnectionHelper.java Mon Apr 8 15:19:04 2013
@@ -0,0 +1,862 @@
+/*
+ *
+ * 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.
+ *
+ */
+package org.apache.qpid.qmf2.util;
+
+// JMS Imports
+import javax.jms.ConnectionFactory;
+import javax.jms.Connection;
+import javax.jms.JMSException;
+
+// JNDI Imports
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
+
+// Simple Logging Facade 4 Java
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+// Misc Imports
+import java.util.Map;
+import java.util.Properties;
+
+// Reuse this class as it provides a handy mechanism to parse an options String into a Map
+import org.apache.qpid.messaging.util.AddressParser;
+
+/**
+ * The Qpid M4 Java and C++ clients and the Python QMF tools all use different URL formats.
+ * This class provides helper methods to support a variety of URL formats and connection options
+ * in order to provide flexibility when creating connections.
+ * <p>
+ * Much of the following information is taken from <a href="https://cwiki.apache.org/qpid/url-format-proposal.html">
+ * New URL format for AMQP + Qpid</a>
+ * <p>
+ * <h3>AMQP 0-10 format</h3>
+ * C++ uses the AMQP 0-10 format: section 9.1.2 as follows:
+ * <pre>
+ * amqp_url = "amqp:" prot_addr_list
+ * prot_addr_list = [prot_addr ","]* prot_addr
+ * prot_addr = tcp_prot_addr | tls_prot_addr
+ *
+ * tcp_prot_addr = tcp_id tcp_addr
+ * tcp_id = "tcp:" | ""
+ * tcp_addr = [host [":" port] ]
+ * host = <as per <a href="http://www.ietf.org/rfc/rfc3986.txt">rfc3986</a>>
+ * port = number
+ * </pre>
+ * The AMQP 0-10 format only provides protocol address information for a (list of) brokers.
+ * <p>
+ * <p>
+ *
+ * <h3>Python tool BrokerURL format</h3>
+ * The Python tools bundled with Qpid such as qpid-config use a "BrokerURL" format with the following Address syntax:
+ * <pre>
+ * [<user>/<pass>@]<hostname> | <ip-address>[:<port>]
+ * </pre>
+ *
+ * <h3>Qpid M4 Java Connection URL format</h3>
+ * The Qpid M4 Java format provides additional options for connection options (user, password, vhost etc.)
+ * Documentation for this format may be found here: <a href="https://cwiki.apache.org/qpid/connection-url-format.html">
+ * Qpid M4 Java Connection URL Format</a>
+ * <p>
+ * Java ConnectionURLs look like this:
+ * <pre>
+ * amqp://[<user>:<pass>@][<clientid>]/<virtualhost>[?<option>='<value>'[&<option>='<value>']]
+ * </pre>
+ * This syntax is very powerful, but it can also be fairly complex to work with, especially when one realises
+ * that one of the options in the above syntax is brokerlist='<broker url>' where broker url is itself a URL
+ * of the format:
+ * <pre>
+ * <transport>://<host>[:<port>][?<option>='<value>'[&<option>='<value>']]
+ * </pre>
+ * so one may see ConnectionURLs that look like:
+ * <pre>
+ * amqp://guest:guest@clientid/test?brokerlist='tcp://localhost:5672?retries='10'&connectdelay='1000''
+ * </pre>
+ *
+ * <p>
+ * <p>
+ * <h3>Extended AMQP 0-10 URL format</h3>
+ * There is a proposal to extend the AMQP 0-10 URL syntax to include user:pass@ style authentication
+ * information, virtual host and extensible name/value options. It also makes the implied extension points of
+ * the original grammar more explicit.
+ * <pre>
+ * amqp_url = "amqp://" [ userinfo "@" ] addr_list [ vhost ]
+ * addr_list = addr *( "," addr )
+ * addr = prot_addr [ options ]
+ * prot_addr = tcp_prot_addr | other_prot_addr
+ * vhost = "/" *pchar [ options ]
+ *
+ * tcp_prot_addr = tcp_id tcp_addr
+ * tcp_id = "tcp:" / "" ; tcp is the default
+ * tcp_addr = [ host [ ":" port ] ]
+ *
+ * other_prot_addr = other_prot_id ":" *pchar
+ * other_prot_id = scheme
+ *
+ * options = "?" option *( ";" option )
+ * option = name "=" value
+ * name = *pchar
+ * value = *pchar
+ * </pre>
+ *
+ * <h3>Incompatibility with AMQP 0-10 format</h3>
+ * This syntax is backward compatible with AMQP 0-10 with one exception: AMQP 0-10 did not have an initial
+ * // after amqp: The justification was that that the // form is only used for URIs with hierarchical structure
+ * <p>
+ * However it's been pointed out that in fact the URL does already specify a 1-level hierarchy of address / vhost.
+ * In the future the hierarchy could be extended to address objects within a vhost such as queues, exchanges etc.
+ * So this proposal adopts amqp:// syntax.
+ * <p>
+ * It's easy to write a backward-compatible parser by relaxing the grammar as follows:
+ * <pre>
+ * amqp_url = "amqp:" [ "//" ] [ userinfo "@" ] addr_list [ vhost ]
+ * </pre>
+ *
+ * <h3>Differences from Qpid M4 Java Connection URL format</h3>
+ * Addresses are at the start of the URL rather than in the "brokerlist" option.
+ * <p>
+ * Option format is ?foo=bar;x=y rather than ?foo='bar'&x='y'. The use of "'" quotes is not common for URI query
+ * strings. The use of "&" as a separator creates problems
+ * <p>
+ * user, pass and clientid are options rather than having a special place at the front of the URL. clientid is
+ * a Qpid proprietary property and user/pass are not relevant in all authentication schemes.
+ * <p>
+ * Qpid M4 Java URLs requires the brokerlist option, so this is an easy way to detect a Qpid M4 URL vs. an
+ * Extended AMQP 0-10 URL and parse accordingly.
+ *
+ * <h3>Options</h3>
+ * Some of the URL forms are fairly limited in terms of options, so it is useful to be able to pass options as
+ * an additional string, though it's important to note that if multiple brokers are supplied in the AMQP 0.10 format
+ * the same options will be applied to all brokers.
+ * <p>
+ * The option format is the same as that of the C++ qpid::messaging Connection class. for example: "{reconnect: true,
+ * tcp-nodelay: true}":
+ * <p>
+ * <table summary="Connection Options" width="100%" border="1"><colgroup><col><col><col></colgroup><thead>
+ * <tr><th>option name</th><th>value type</th><th>semantics</th></tr></thead><tbody>
+ * <tr>
+ * <td><code class="literal">maxprefetch</code></td>
+ * <td>integer</td>
+ * <td>The maximum number of pre-fetched messages per destination.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">sync_publish</code></td>
+ * <td>{'persistent' | 'all'}</td>
+ * <td>A sync command is sent after every persistent message to guarantee that it has been received; if the
+ * value is 'persistent', this is done only for persistent messages.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">sync_ack</code></td>
+ * <td>boolean</td>
+ * <td>A sync command is sent after every acknowledgement to guarantee that it has been received.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">use_legacy_map_msg_format</code></td>
+ * <td>boolean</td>
+ * <td>If you are using JMS Map messages and deploying a new client with any JMS client older than 0.8 release,
+ * you must set this to true to ensure the older clients can understand the map message encoding.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">failover</code></td>
+ * <td>{'roundrobin' | 'singlebroker' | 'nofailover' | 'failover_exchange'}</td>
+ * <td>If roundrobin is selected it will try each broker given in the broker list. If failover_exchange is
+ * selected it connects to the initial broker given in the broker URL and will receive membership updates
+ * via the failover exchange. </td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">cyclecount</code></td>
+ * <td>integer</td>
+ * <td>For roundrobin failover cyclecount is the number of times to loop through the list of available brokers
+ * before failure.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">username</code></td>
+ * <td>string</td>
+ * <td>The username to use when authenticating to the broker.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">password</code></td>
+ * <td>string</td>
+ * <td>The password to use when authenticating to the broker.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">sasl_mechanisms</code></td>
+ * <td>string</td>
+ * <td>The specific SASL mechanisms to use when authenticating to the broker. The value is a space separated list.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">sasl_mechs</code></td>
+ * <td>string</td>
+ * <td>The specific SASL mechanisms to use when authenticating to the broker. The value is a space separated
+ * is a space separated list. This is simply a synonym for sasl_mechanisms above</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">sasl_encryption</code></td>
+ * <td>boolean</td>
+ * <td>If <code class="literal">sasl_encryption='true'</code>, the JMS client attempts to negotiate a security
+ * layer with the broker using GSSAPI to encrypt the connection. Note that for this to happen, GSSAPI must
+ * be selected as the sasl_mech.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">ssl</code></td>
+ * <td>boolean</td>
+ * <td>If <code class="literal">ssl='true'</code>, the JMS client will encrypt the connection using SSL.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">reconnect</code></td>
+ * <td>boolean</td>
+ * <td>Transparently reconnect if the connection is lost.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">reconnect_timeout</code></td>
+ * <td>integer</td>
+ * <td>Total number of seconds to continue reconnection attempts before giving up and raising an exception.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">reconnect_limit</code></td>
+ * <td>integer</td>
+ * <td>Maximum number of reconnection attempts before giving up and raising an exception.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">reconnect_interval_min</code></td>
+ * <td>integer representing time in seconds</td>
+ * <td> Minimum number of seconds between reconnection attempts. The first reconnection attempt is made
+ * immediately; if that fails, the first reconnection delay is set to the value of <code class="literal">
+ * reconnect_interval_min</code>; if that attempt fails, the reconnect interval increases exponentially
+ * until a reconnection attempt succeeds or <code class="literal">reconnect_interval_max</code> is reached.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">reconnect_interval_max</code></td>
+ * <td>integer representing time in seconds</td>
+ * <td>Maximum reconnect interval.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">reconnect_interval</code></td>
+ * <td>integer representing time in seconds</td>
+ * <td>Sets both <code class="literal">reconnection_interval_min</code> and <code class="literal">
+ * reconnection_interval_max</code> to the same value. The default value is 5 seconds</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">heartbeat</code></td>
+ * <td>integer representing time in seconds</td>
+ * <td>Requests that heartbeats be sent every N seconds. If two successive heartbeats are missed the connection is
+ * considered to be lost.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">protocol</code></td>
+ * <td>string</td>
+ * <td>Sets the underlying protocol used. The default option is 'tcp'. To enable ssl, set to 'ssl'. The C++ client
+ * additionally supports 'rdma'. </td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">tcp-nodelay</code></td>
+ * <td>boolean</td>
+ * <td>Set tcp no-delay, i.e. disable Nagle algorithm.</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">sasl_protocol</code></td>
+ * <td>string</td>
+ * <td>Used only for Kerberos. <code class="literal">sasl_protocol</code> must be set to the principal for the
+ * qpidd broker, e.g. <code class="literal">qpidd/</code></td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">sasl_server</code></td>
+ * <td>string</td>
+ * <td>For Kerberos, sasl_mechs must be set to GSSAPI, <code class="literal">sasl_server</code> must be set to
+ * the host for the SASL server, e.g. <code class="literal">sasl.com.</code></td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">trust_store</code></td>
+ * <td>string</td>
+ * <td>path to Keberos trust store</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">trust_store_password</code></td>
+ * <td>string</td>
+ * <td>Kerberos trust store password</td>
+ * </tr>
+ * <tr>
+ * <td><code class="key_store</code></td>
+ * <td>string</td>
+ * <td>path to Kerberos key store </td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">key_store_password</code></td>
+ * <td>string</td>
+ * <td>Kerberos key store password</td>
+ * </tr>
+ * <tr>
+ * <td><code class="literal">ssl_cert_alias</code></td>
+ * <td>string</td>
+ * <td>If multiple certificates are present in the keystore, the alias will be used to extract the correct
+ * certificate.</td>
+ * </tr>
+ * </tbody></table>
+ *
+ * <h3>Other features of this class</h3>
+ * Whilst it is generally the norm to use JNDI to specify Connections, Destinations etc. it is also often quite useful
+ * to specify Connections programmatically, for example when writing a tool one may wish to specify the broker via the
+ * command line to enable the tool to connect to different broker instances.
+ * To facilitate this this class provides a basic createConnection() method that takes a URL and returns a JMS
+ * Connection.
+ *
+ * @author Fraser Adams
+ */
+public final class ConnectionHelper
+{
+ private static final Logger _log = LoggerFactory.getLogger(ConnectionHelper.class);
+
+ /**
+ * Make constructor private as the class comprises only static helper methods.
+ */
+ private ConnectionHelper()
+ {
+ }
+
+ /**
+ * Create a ConnectionURL from the proposed Extended AMQP 0.10 URL format. This is experimental and may or may
+ * not work. Options are assumed to be the same as the Java Connection URL, which will probably not be the case
+ * if this URL form is ultimately adopted. For example the example URLs have "amqp://host1,host2?retry=2,host3"
+ * whereas the Java Connection URL uses &retries=2
+ *
+ * I'm not overly keen on this code it looks pretty inelegant and I'm slightly embarrassed by it, but it
+ * is really just an experiment.
+ *
+ * @param url the input URL.
+ * @param username the username.
+ * @param password the password.
+ * @return a String containing the Java Connection URL.
+ */
+ private static String parseExtendedAMQPURL(String url, String username, String password)
+ {
+ String vhost = ""; // Specifying an empty vhost uses default Virtual Host.
+ String urlOptions = "";
+ String brokerList = "";
+
+ url = url.substring(7); // Chop off "amqp://"
+ String[] split = url.split("@"); // First break out the userinfo if present
+ String remainder = split[0];
+ if (split.length == 2)
+ { // Extract username and password from the userinfo field
+ String[] userinfo = split[0].split(":");
+ remainder = split[1];
+
+ username = userinfo[0];
+ if (userinfo.length == 2)
+ {
+ password = userinfo[1];
+ }
+ }
+
+ // Replace foo=baz with foo='baz'. There's probably a more elegant way to do this using a fancy
+ // regex, but unfortunately I'm not terribly good at regexes so this is the brute force approach :-(
+ // OTOH it's probably more readable and obvious than a regex to do the same thing would be.
+ split = remainder.split("=");
+ StringBuilder buf = new StringBuilder(split[0]);
+ for (int i = 1; i < split.length; i++)
+ {
+ String substring = "='" + split[i];
+ if (substring.contains(";"))
+ {
+ substring = substring.replaceFirst(";", "'&"); // Note we also replace the option separator here
+ }
+ else if (substring.contains("/"))
+ {
+ substring = substring.replaceFirst("/", "'/");
+ }
+ else if (substring.contains(","))
+ {
+ substring = substring.replaceFirst(",", "',");
+ }
+ else
+ {
+ substring = substring + "'";
+ }
+ buf.append(substring);
+ }
+ remainder = buf.toString();
+
+ // Now split into addrList and vhost parts (see Javadoc for the grammar of this URL format)
+ split = remainder.split("/"); // vhost starts with a mandatory '/' character
+ String[] addrSplit = split[0].split(","); // prot_addrs are comma separated
+ boolean firstBroker = true;
+ buf = new StringBuilder();
+ for (String broker : addrSplit)
+ { // Iterate through the address list creating brokerList style URLs
+ broker = broker.trim();
+ String protocol = "tcp"; // set default protocol
+ String[] components = broker.split(":");
+
+ // Note protocols other than tcp and vm are not supported by the Connection URL so the results
+ // are pretty much undefined if other protocols are passed on the input URL.
+ if (components.length == 1)
+ { // Assume protocol = tcp and broker = hostname
+ }
+ else if (components.length == 2)
+ { // Probably host:port but could be some other protocol in and Extended AMQP 0.10 URL
+ try
+ { // Somewhat ugly, but effective test to check if the second component is an integer
+ Integer.parseInt(components[1]);
+ // If the above succeeds the components are likely host:port
+ broker = components[0] + ":" + components[1];
+ }
+ catch (NumberFormatException nfe)
+ { // If the second component isn't an integer then it's likely a wacky protocol...
+ protocol = components[0];
+ broker = components[1];
+ }
+ }
+ else if (components.length == 3)
+ {
+ protocol = components[0];
+ broker = components[1] + ":" + components[2];
+ }
+
+ if (firstBroker)
+ {
+ buf.append(protocol + "://" + broker);
+ }
+ else
+ { // https://cwiki.apache.org/qpid/connection-url-format.html says "A minimum of one broker url is
+ // required additional URLs are semi-colon(';') delimited."
+ buf.append(";" + protocol + "://" + broker);
+ }
+ firstBroker = false;
+ }
+ brokerList = "'" + buf.toString() + "'";
+
+ if (split.length == 2)
+ { // Extract the vhost and any connection level options
+ vhost = split[1];
+ String[] split2 = vhost.split("\\?"); // Look for options
+ vhost = split2[0];
+ if (split2.length == 2)
+ {
+ urlOptions = "&" + split2[1];
+ }
+ }
+
+ String connectionURL = "amqp://" + username + ":" + password + "@QpidJMS/" + vhost + "?brokerlist=" +
+ brokerList + urlOptions;
+ return connectionURL;
+ }
+
+ /**
+ * If no explicit username is supplied then explicitly set sasl mechanism to ANONYMOUS. If this isn't done
+ * The default is PLAIN which causes the broker to fail with "warning Failed to retrieve sasl username".
+ *
+ * @param username the previously extracted username.
+ * @param brokerListOptions the brokerList options extracted so far.
+ * @return the brokerList options adjusted with sasl_mechs='ANONYMOUS' if no username has been supplied.
+ */
+ private static String adjustBrokerListOptions(final String username, final String brokerListOptions)
+ {
+ if (username.equals(""))
+ {
+ if (brokerListOptions.equals(""))
+ {
+ return "?sasl_mechs='ANONYMOUS'";
+ }
+ else
+ {
+ if (brokerListOptions.contains("sasl_mechs"))
+ {
+ return brokerListOptions;
+ }
+ else
+ {
+ return brokerListOptions + "&sasl_mechs='ANONYMOUS'";
+ }
+ }
+ }
+ else
+ {
+ return brokerListOptions;
+ }
+ }
+
+ /**
+ * Create a ConnectionURL from the input "generic" URL.
+ *
+ * @param url the input URL.
+ * @param username the username.
+ * @param password the password.
+ * @param urlOptions the pre-parsed set of connection level options.
+ * @param brokerListOptions the pre-parsed set of specific brokerList options.
+ * @return a String containing the Java Connection URL.
+ */
+ private static String parseURL(String url, String username, String password,
+ String urlOptions, String brokerListOptions)
+ {
+ if (url.startsWith("amqp://"))
+ { // Somewhat experimental. This new format is only a "proposed" format
+ return parseExtendedAMQPURL(url, username, password);
+ }
+
+ String vhost = ""; // Specifying an empty vhost uses default Virtual Host.
+ String brokerList = "";
+ if (url.startsWith("amqp:"))
+ { // AMQP 0.10 URL format
+ url = url.substring(5); // Chop off "amqp:"
+ String[] addrSplit = url.split(","); // prot_addrs are comma separated
+ boolean firstBroker = true;
+ brokerListOptions = adjustBrokerListOptions(username, brokerListOptions);
+ StringBuilder buf = new StringBuilder();
+ for (String broker : addrSplit)
+ { // Iterate through the address list creating brokerList style URLs
+ broker = broker.trim();
+ if (broker.startsWith("tcp:"))
+ { // Only tcp is supported in an AMQP 0.10 prot_addr so we *should* only have to account for
+ // a "tcp:" prefix when normalising broker URLs
+ broker = broker.substring(4); // Chop off "tcp:"
+ }
+
+ if (firstBroker)
+ {
+ buf.append("tcp://" + broker + brokerListOptions);
+ }
+ else
+ { // https://cwiki.apache.org/qpid/connection-url-format.html says "A minimum of one broker url is
+ // required additional URLs are semi-colon(';') delimited."
+ buf.append(";tcp://" + broker + brokerListOptions);
+ }
+ firstBroker = false;
+ }
+ brokerList = "'" + buf.toString() + "'";
+ }
+ else if (url.contains("@"))
+ { // BrokerURL format
+ String[] split = url.split("@");
+ url = split[1];
+
+ split = split[0].split("/");
+ username = split[0];
+
+ if (split.length == 2)
+ {
+ password = split[1];
+ }
+
+ brokerListOptions = adjustBrokerListOptions(username, brokerListOptions);
+ brokerList = "'tcp://" + url + brokerListOptions + "'";
+ }
+ else
+ { // Basic host:port format
+ brokerListOptions = adjustBrokerListOptions(username, brokerListOptions);
+ brokerList = "'tcp://" + url + brokerListOptions + "'";
+ }
+
+ String connectionURL = "amqp://" + username + ":" + password + "@QpidJMS/" + vhost + "?brokerlist=" +
+ brokerList + urlOptions;
+ return connectionURL;
+ }
+
+
+ /**
+ * Creates a Java Connection URL from one of the other supported URL formats.
+ *
+ * @param url an AMQP 0.10 URL, an extended AMQP 0-10 URL, a Broker URL or a Connection URL (the latter is simply
+ * returned untouched).
+ * @return a String containing the Java Connection URL.
+ */
+ public static String createConnectionURL(String url)
+ {
+ return createConnectionURL(url, null);
+ }
+
+ /**
+ * Creates a Java Connection URL from one of the other supported URL formats plus options.
+ *
+ * @param url an AMQP 0.10 URL, an extended AMQP 0-10 URL, a Broker URL or a Connection URL (the latter is simply
+ * returned untouched).
+ * @param opts a String containing the options encoded using the same form as the C++ qpid::messaging
+ * Connection class.
+ * @return a String containing the Java Connection URL.
+ */
+ public static String createConnectionURL(String url, String opts)
+ {
+ // This method is actually mostly about parsing the options, when the options are extracted it delegates
+ // to parseURL() to do the actual URL parsing.
+
+ // If a Java Connection URL has been passed in we simply return it.
+ if (url.startsWith("amqp://") && url.contains("brokerlist"))
+ {
+ return url;
+ }
+
+ // Initialise options to default values
+ String username = "";
+ String password = "";
+ String urlOptions = "";
+ String brokerListOptions = "";
+
+ // Get options from option String
+ if (opts != null && opts.startsWith("{") && opts.endsWith("}"))
+ {
+ // Connection URL Options
+ String maxprefetch = "";
+ String sync_publish = "";
+ String sync_ack = "";
+ String use_legacy_map_msg_format = "";
+ String failover = "";
+
+ // Broker List Options
+ String heartbeat = "";
+ String retries = "";
+ String connectdelay = "";
+ String connecttimeout = "";
+
+ String tcp_nodelay = "";
+
+ String sasl_mechs = "";
+ String sasl_encryption = "";
+ String sasl_protocol = "";
+ String sasl_server = "";
+
+ String ssl = "";
+ String ssl_verify_hostname = "";
+ String ssl_cert_alias = "";
+
+ String trust_store = "";
+ String trust_store_password = "";
+ String key_store = "";
+ String key_store_password = "";
+
+ Map options = new AddressParser(opts).map();
+
+ if (options.containsKey("maxprefetch"))
+ {
+ maxprefetch = "&maxprefetch='" + options.get("maxprefetch").toString() + "'";
+ }
+
+ if (options.containsKey("sync_publish"))
+ {
+ sync_publish = "&sync_publish='" + options.get("sync_publish").toString() + "'";
+ }
+
+ if (options.containsKey("sync_ack"))
+ {
+ sync_ack = "&sync_ack='" + options.get("sync_ack").toString() + "'";
+ }
+
+ if (options.containsKey("use_legacy_map_msg_format"))
+ {
+ use_legacy_map_msg_format = "&use_legacy_map_msg_format='" +
+ options.get("use_legacy_map_msg_format").toString() + "'";
+ }
+
+ if (options.containsKey("failover"))
+ {
+ if (options.containsKey("cyclecount"))
+ {
+ failover = "&failover='" + options.get("failover").toString() + "?cyclecount='" +
+ options.get("cyclecount").toString() + "''";
+ }
+ else
+ {
+ failover = "&failover='" + options.get("failover").toString() + "'";
+ }
+ }
+
+ if (options.containsKey("username"))
+ {
+ username = options.get("username").toString();
+ }
+
+ if (options.containsKey("password"))
+ {
+ password = options.get("password").toString();
+ }
+
+ if (options.containsKey("reconnect"))
+ {
+ String value = options.get("reconnect").toString();
+ if (value.equalsIgnoreCase("true"))
+ {
+ retries = "&retries='" + Integer.MAX_VALUE + "'";
+ connectdelay = "&connectdelay='5000'";
+ }
+ }
+
+ if (options.containsKey("reconnect_limit"))
+ {
+ retries = "&retries='" + options.get("reconnect_limit").toString() + "'";
+ }
+
+ if (options.containsKey("reconnect_interval"))
+ {
+ connectdelay = "&connectdelay='" + options.get("reconnect_interval").toString() + "000'";
+ }
+
+ if (options.containsKey("reconnect_interval_min"))
+ {
+ connectdelay = "&connectdelay='" + options.get("reconnect_interval_min").toString() + "000'";
+ }
+
+ if (options.containsKey("reconnect_interval_max"))
+ {
+ connectdelay = "&connectdelay='" + options.get("reconnect_interval_max").toString() + "000'";
+ }
+
+ if (options.containsKey("reconnect_timeout"))
+ {
+ connecttimeout = "&connecttimeout='" + options.get("reconnect_timeout").toString() + "000'";
+ }
+
+ if (options.containsKey("heartbeat"))
+ {
+ heartbeat = "&heartbeat='" + options.get("heartbeat").toString() + "'";
+ }
+
+ if (options.containsKey("tcp-nodelay"))
+ {
+ tcp_nodelay = "&tcp_nodelay='" + options.get("tcp-nodelay").toString() + "'";
+ }
+
+ if (options.containsKey("sasl_mechanisms"))
+ {
+ sasl_mechs = "&sasl_mechs='" + options.get("sasl_mechanisms").toString() + "'";
+ }
+
+ if (options.containsKey("sasl_mechs"))
+ {
+ sasl_mechs = "&sasl_mechs='" + options.get("sasl_mechs").toString() + "'";
+ }
+
+ if (options.containsKey("sasl_encryption"))
+ {
+ sasl_encryption = "&sasl_encryption='" + options.get("sasl_encryption").toString() + "'";
+ }
+
+ if (options.containsKey("sasl_protocol"))
+ {
+ sasl_protocol = "&sasl_protocol='" + options.get("sasl_protocol").toString() + "'";
+ }
+
+ if (options.containsKey("sasl_server"))
+ {
+ sasl_server = "&sasl_server='" + options.get("sasl_server").toString() + "'";
+ }
+
+ if (options.containsKey("trust_store"))
+ {
+ trust_store = "&trust_store='" + options.get("trust_store").toString() + "'";
+ }
+
+ if (options.containsKey("trust_store_password"))
+ {
+ trust_store_password = "&trust_store_password='" + options.get("trust_store_password").toString() + "'";
+ }
+
+ if (options.containsKey("key_store"))
+ {
+ key_store = "&key_store='" + options.get("key_store").toString() + "'";
+ }
+
+ if (options.containsKey("key_store_password"))
+ {
+ key_store_password = "&key_store_password='" + options.get("key_store_password").toString() + "'";
+ }
+
+ if (options.containsKey("protocol"))
+ {
+ String value = options.get("protocol").toString();
+ if (value.equalsIgnoreCase("ssl"))
+ {
+ ssl = "&ssl='true'";
+ if (options.containsKey("ssl_verify_hostname"))
+ {
+ ssl_verify_hostname = "&ssl_verify_hostname='" + options.get("ssl_verify_hostname").toString() + "'";
+ }
+
+ if (options.containsKey("ssl_cert_alias"))
+ {
+ ssl_cert_alias = "&ssl_cert_alias='" + options.get("ssl_cert_alias").toString() + "'";
+ }
+ }
+ }
+
+ urlOptions = maxprefetch + sync_publish + sync_ack + use_legacy_map_msg_format + failover;
+
+ brokerListOptions = heartbeat + retries + connectdelay + connecttimeout + tcp_nodelay +
+ sasl_mechs + sasl_encryption + sasl_protocol + sasl_server +
+ ssl + ssl_verify_hostname + ssl_cert_alias +
+ trust_store + trust_store_password + key_store + key_store_password;
+
+ if (brokerListOptions.startsWith("&"))
+ {
+ brokerListOptions = "?" + brokerListOptions.substring(1);
+ }
+ }
+
+ return parseURL(url, username, password, urlOptions, brokerListOptions);
+ }
+
+ /**
+ * Creates a JMS Connection from one of the supported URL formats.
+ *
+ * @param url an AMQP 0.10 URL, an extended AMQP 0-10 URL, a Broker URL or a Connection URL.
+ * @return a javax.jms.Connection.
+ */
+ public static Connection createConnection(String url)
+ {
+ return createConnection(url, null);
+ }
+
+ /**
+ * Creates a JMS Connection from one of the supported URL formats plus options.
+ *
+ * @param url an AMQP 0.10 URL, an extended AMQP 0-10 URL, a Broker URL or a Connection URL.
+ * @param opts a String containing the options encoded using the same form as the C++ qpid::messaging
+ * Connection class.
+ * @return a javax.jms.Connection.
+ */
+ public static Connection createConnection(String url, String opts)
+ {
+ String connectionUrl = createConnectionURL(url, opts);
+ _log.info("ConnectionHelper.createConnection() {}", connectionUrl);
+
+ // Initialise JNDI names etc into properties
+ Properties props = new Properties();
+ props.setProperty("java.naming.factory.initial", "org.apache.qpid.jndi.PropertiesFileInitialContextFactory");
+ props.setProperty("connectionfactory.ConnectionFactory", connectionUrl);
+
+ Connection connection = null;
+ try
+ {
+ Context jndi = new InitialContext(props);
+ ConnectionFactory connectionFactory = (ConnectionFactory)jndi.lookup("ConnectionFactory");
+ connection = connectionFactory.createConnection();
+ }
+ catch (NamingException ne)
+ {
+ _log.info("NamingException {} caught in createConnection()", ne.getMessage());
+ }
+ catch (JMSException jmse)
+ {
+ _log.info("JMSException {} caught in createConnection()", jmse.getMessage());
+ }
+
+ return connection;
+ }
+}
+
Propchange: qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/ConnectionHelper.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/ConnectionHelper.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/GetOpt.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/GetOpt.java?rev=1465662&view=auto
==============================================================================
--- qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/GetOpt.java (added)
+++ qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/GetOpt.java Mon Apr 8 15:19:04 2013
@@ -0,0 +1,208 @@
+/*
+ *
+ * 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.
+ *
+ */
+package org.apache.qpid.qmf2.util;
+
+// Misc Imports
+import java.util.List;
+import java.util.ArrayList;
+
+/**
+ * Basic Java port of the python getopt function.
+ *
+ * Takes a standard args String array plus short form and long form options lists.
+ * Searches the args for options and any option arguments and stores these in optList
+ * any remaining args are stored in encArgs.
+ *
+ * <p>
+ * Example usage (paraphrased from QpidConfig):
+ * <pre>
+ * String[] longOpts = {"help", "durable", "cluster-durable", "bindings", "broker-addr=", "file-count=",
+ * "file-size=", "max-queue-size=", "max-queue-count=", "limit-policy=",
+ * "order=", "sequence", "ive", "generate-queue-events=", "force", "force-if-not-empty",
+ * "force-if-used", "alternate-exchange=", "passive", "timeout=", "file=", "flow-stop-size=",
+ * "flow-resume-size=", "flow-stop-count=", "flow-resume-count=", "argument="};
+ *
+ * try
+ * {
+ * GetOpt getopt = new GetOpt(args, "ha:bf:", longOpts);
+ * List<String[]> optList = getopt.getOptList();
+ * String[] cargs = {};
+ * cargs = getopt.getEncArgs().toArray(cargs);
+ *
+ * for (String[] opt : optList)
+ * {
+ * //System.out.println(opt[0] + ":" + opt[1]);
+ * if (opt[0].equals("-a") || opt[0].equals("--broker-addr"))
+ * {
+ * _host = opt[1];
+ * }
+ * // Just a sample - more parsing would follow....
+ * }
+ *
+ * int nargs = cargs.length;
+ * if (nargs == 0)
+ * {
+ * overview();
+ * }
+ * else
+ * {
+ * String cmd = cargs[0];
+ * String modifier = "";
+ *
+ * if (nargs > 1)
+ * {
+ * modifier = cargs[1];
+ * }
+ *
+ * if (cmd.equals("exchanges"))
+ * {
+ * if (_recursive)
+ * {
+ * exchangeListRecurse(modifier);
+ * }
+ * else
+ * {
+ * exchangeList(modifier);
+ * }
+ * }
+ * // Just a sample - more parsing would follow....
+ * }
+ * }
+ * catch (IllegalArgumentException e)
+ * {
+ * System.err.println(e.toString());
+ * usage();
+ * }
+ * </pre>
+ *
+ * @author Fraser Adams
+ */
+public final class GetOpt
+{
+ private List<String[]> _optList = new ArrayList<String[]>();
+ private List<String> _encArgs = new ArrayList<String>();
+
+ /**
+ * Returns the options and option arguments as a list containing a String array. The first element of the array
+ * contains the option and the second contains any arguments if present.
+ *
+ * @return the options and option arguments
+ */
+ public List<String[]> getOptList()
+ {
+ return _optList;
+ }
+
+ /**
+ * Returns any remaining arguments. This is any argument from a command line that doesn't begin "--" or "-".
+ *
+ * @return any remaining arguments not made available by getOptList().
+ */
+ public List<String> getEncArgs()
+ {
+ return _encArgs;
+ }
+
+ /**
+ * Takes a standard args String array plus short form and long form options lists.
+ * Searches the args for options and any option arguments and stores these in optList
+ * any remaining args are stored in encArgs.
+ *
+ * @param args standard arg String array
+ * @param opts short form option list of the form "ab:cd:" where b and d are options with arguments.
+ * @param longOpts long form option list as an array of strings. "option=" signifies an options with arguments.
+ */
+ public GetOpt(String[] args, String opts, String[] longOpts) throws IllegalArgumentException
+ {
+ int argslength = args.length;
+ for (int i = 0; i < argslength; i++)
+ {
+ String arg = args[i];
+ if (arg.startsWith("--"))
+ {
+ String extractedOption = arg.substring(2);
+ int nargs = _optList.size();
+ for (String j : longOpts)
+ {
+ String k = j.substring(0, j.length() - 1);
+ if (j.equals(extractedOption) || k.equals(extractedOption))
+ { // Handle where option and value are space separated
+ String arg0 = arg;
+ String arg1 = "";
+ if (i < argslength - 1 && k.equals(extractedOption))
+ {
+ i++;
+ arg1 = args[i];
+ }
+ String[] option = {arg0, arg1};
+ _optList.add(option);
+ }
+ else
+ { // Handle where option and value are separated by '='
+ String[] split = arg.split("=", 2); // Split on the first occurrence of '='
+ String arg0 = split[0];
+ String arg1 = "";
+ if (split.length == 2)
+ {
+ j = j.substring(0, j.length() - 1);
+ arg1 = split[1];
+ }
+
+ if (arg0.substring(2).equals(j))
+ {
+ String[] option = {arg0, arg1};
+ _optList.add(option);
+ }
+ }
+ }
+
+ if (nargs == _optList.size())
+ {
+ throw new IllegalArgumentException("Unknown Option " + arg);
+ }
+ }
+ else if (arg.startsWith("-"))
+ {
+ String extractedOption = arg.substring(1);
+ int index = opts.indexOf(extractedOption);
+ if (index++ != -1)
+ {
+ String arg1 = "";
+ if (i < argslength - 1 && index < opts.length() && opts.charAt(index) == ':')
+ {
+ i++;
+ arg1 = args[i];
+ }
+ String[] option = {arg, arg1};
+ _optList.add(option);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Unknown Option " + arg);
+ }
+ }
+ else
+ {
+ _encArgs.add(arg);
+ }
+ }
+ }
+}
Propchange: qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/GetOpt.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/util/GetOpt.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: qpid/trunk/qpid/tools/src/java/src/patch/java/AMQMessageDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/tools/src/java/src/patch/java/AMQMessageDelegate_0_10.java?rev=1465662&view=auto
==============================================================================
--- qpid/trunk/qpid/tools/src/java/src/patch/java/AMQMessageDelegate_0_10.java (added)
+++ qpid/trunk/qpid/tools/src/java/src/patch/java/AMQMessageDelegate_0_10.java Mon Apr 8 15:19:04 2013
@@ -0,0 +1,958 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+package org.apache.qpid.client.message;
+
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageFormatException;
+import javax.jms.MessageNotWriteableException;
+import javax.jms.Session;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQPInvalidClassException;
+import org.apache.qpid.collections.ReferenceMap;
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.AMQSession_0_10;
+import org.apache.qpid.client.CustomJMSXProperty;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.jms.Message;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.ExchangeQueryResult;
+import org.apache.qpid.transport.Future;
+import org.apache.qpid.transport.Header;
+import org.apache.qpid.transport.MessageDeliveryMode;
+import org.apache.qpid.transport.MessageDeliveryPriority;
+import org.apache.qpid.transport.MessageProperties;
+import org.apache.qpid.transport.ReplyTo;
+
+/**
+ * This extends AbstractAMQMessageDelegate which contains common code between
+ * both the 0_8 and 0_10 Message types.
+ *
+ */
+public class AMQMessageDelegate_0_10 extends AbstractAMQMessageDelegate
+{
+ private static final Map<ReplyTo, Destination> _destinationCache = Collections.synchronizedMap(new ReferenceMap());
+
+ public static final String JMS_TYPE = "x-jms-type";
+
+
+ private boolean _readableProperties = false;
+
+ private Destination _destination;
+
+
+ private MessageProperties _messageProps;
+ private DeliveryProperties _deliveryProps;
+ /** If the acknowledge mode is CLIENT_ACKNOWLEDGE the session is required */
+ private AMQSession _session;
+ private final long _deliveryTag;
+
+
+ protected AMQMessageDelegate_0_10()
+ {
+ this(new MessageProperties(), new DeliveryProperties(), -1);
+ _readableProperties = false;
+ }
+
+ protected AMQMessageDelegate_0_10(MessageProperties messageProps, DeliveryProperties deliveryProps, long deliveryTag)
+ {
+ _messageProps = messageProps;
+ _deliveryProps = deliveryProps;
+ _deliveryTag = deliveryTag;
+ _readableProperties = (_messageProps != null);
+
+ AMQDestination dest;
+
+ dest = generateDestination(new AMQShortString(_deliveryProps.getExchange()),
+ new AMQShortString(_deliveryProps.getRoutingKey()));
+ setJMSDestination(dest);
+ }
+
+ /**
+ * Use the 0-10 ExchangeQuery call to validate the exchange type.
+ *
+ * This is used primarily to provide the correct JMSDestination value.
+ *
+ * The query is performed synchronously iff the map exchange is not already
+ * present in the exchange Map.
+ *
+ * @param header The message headers, from which the exchange name can be extracted
+ * @param session The 0-10 session to use to call ExchangeQuery
+ */
+ public static void updateExchangeTypeMapping(Header header, org.apache.qpid.transport.Session session)
+ {
+ DeliveryProperties deliveryProps = header.get(DeliveryProperties.class);
+ if (deliveryProps != null)
+ {
+ String exchange = deliveryProps.getExchange();
+
+ if (exchange != null && !exchangeMapContains(exchange))
+ {
+ Future<ExchangeQueryResult> future =
+ session.exchangeQuery(exchange.toString());
+ ExchangeQueryResult res = future.get();
+
+ updateExchangeType(exchange, res.getType());
+ }
+ }
+ }
+
+
+ public String getJMSMessageID() throws JMSException
+ {
+ UUID id = _messageProps.getMessageId();
+ return id == null ? null : "ID:" + id;
+ }
+
+ public void setJMSMessageID(String messageId) throws JMSException
+ {
+ if(messageId == null)
+ {
+ _messageProps.clearMessageId();
+ }
+ else
+ {
+ if(messageId.startsWith("ID:"))
+ {
+ try
+ {
+ _messageProps.setMessageId(UUID.fromString(messageId.substring(3)));
+ }
+ catch(IllegalArgumentException ex)
+ {
+ throw new JMSException("MessageId '"+messageId+"' is not of the correct format, it must be ID: followed by a UUID");
+ }
+ }
+ else
+ {
+ throw new JMSException("MessageId '"+messageId+"' is not of the correct format, it must be ID: followed by a UUID");
+ }
+ }
+ }
+
+ public void setJMSMessageID(UUID messageId) throws JMSException
+ {
+ if(messageId == null)
+ {
+ _messageProps.clearMessageId();
+ }
+ else
+ {
+ _messageProps.setMessageId(messageId);
+ }
+ }
+
+
+ public long getJMSTimestamp() throws JMSException
+ {
+ return _deliveryProps.getTimestamp();
+ }
+
+ public void setJMSTimestamp(long timestamp) throws JMSException
+ {
+ _deliveryProps.setTimestamp(timestamp);
+ }
+
+ public byte[] getJMSCorrelationIDAsBytes() throws JMSException
+ {
+ return _messageProps.getCorrelationId();
+ }
+
+ public void setJMSCorrelationIDAsBytes(byte[] bytes) throws JMSException
+ {
+ _messageProps.setCorrelationId(bytes);
+ }
+
+ public void setJMSCorrelationID(String correlationId) throws JMSException
+ {
+
+ setJMSCorrelationIDAsBytes(correlationId == null ? null : correlationId.getBytes());
+ }
+
+ public String getJMSCorrelationID() throws JMSException
+ {
+
+ byte[] correlationIDAsBytes = getJMSCorrelationIDAsBytes();
+ return correlationIDAsBytes == null ? null : new String(correlationIDAsBytes);
+ }
+
+ public Destination getJMSReplyTo()
+ {
+ ReplyTo replyTo = _messageProps.getReplyTo();
+
+ if (replyTo == null)
+ {
+ return null;
+ }
+ else
+ {
+ Destination dest = _destinationCache.get(replyTo);
+ if (dest == null)
+ {
+ String exchange = replyTo.getExchange();
+ String routingKey = replyTo.getRoutingKey();
+
+ dest = generateDestination(exchange == null ? null : new AMQShortString(exchange),
+ routingKey == null ? null : new AMQShortString(routingKey));
+
+
+
+
+
+ _destinationCache.put(replyTo, dest);
+ }
+
+ return dest;
+ }
+ }
+
+ public void setJMSReplyTo(Destination destination) throws JMSException
+ {
+ if (destination == null)
+ {
+ _messageProps.setReplyTo(null);
+ return;
+ }
+
+ if (!(destination instanceof AMQDestination))
+ {
+ throw new IllegalArgumentException(
+ "ReplyTo destination may only be an AMQDestination - passed argument was type " + destination.getClass());
+ }
+
+ final AMQDestination amqd = (AMQDestination) destination;
+
+ if (amqd.getDestSyntax() == AMQDestination.DestSyntax.ADDR)
+ {
+ try
+ {
+ int type = ((AMQSession_0_10)_session).resolveAddressType(amqd);
+ if (type == AMQDestination.QUEUE_TYPE)
+ {
+ ((AMQSession_0_10)_session).setLegacyFiledsForQueueType(amqd);
+ }
+ else
+ {
+ ((AMQSession_0_10)_session).setLegacyFiledsForTopicType(amqd);
+ }
+ }
+ catch(AMQException ex)
+ {
+ JMSException e = new JMSException("Error occured while figuring out the node type");
+ e.initCause(ex);
+ e.setLinkedException(ex);
+ throw e;
+ }
+ }
+
+ final ReplyTo replyTo = new ReplyTo(amqd.getExchangeName().toString(), amqd.getRoutingKey().toString());
+ _destinationCache.put(replyTo, destination);
+ _messageProps.setReplyTo(replyTo);
+
+ }
+
+ public Destination getJMSDestination() throws JMSException
+ {
+ return _destination;
+ }
+
+ public void setJMSDestination(Destination destination)
+ {
+ _destination = destination;
+ }
+
+ public void setContentType(String contentType)
+ {
+ _messageProps.setContentType(contentType);
+ }
+
+ public String getContentType()
+ {
+ return _messageProps.getContentType();
+ }
+
+ public void setEncoding(String encoding)
+ {
+ if(encoding == null || encoding.length() == 0)
+ {
+ _messageProps.clearContentEncoding();
+ }
+ else
+ {
+ _messageProps.setContentEncoding(encoding);
+ }
+ }
+
+ public String getEncoding()
+ {
+ return _messageProps.getContentEncoding();
+ }
+
+ public String getReplyToString()
+ {
+ Destination replyTo = getJMSReplyTo();
+ if(replyTo != null)
+ {
+ return ((AMQDestination)replyTo).toURL();
+ }
+ else
+ {
+ return null;
+ }
+
+ }
+
+ public int getJMSDeliveryMode() throws JMSException
+ {
+
+ MessageDeliveryMode deliveryMode = _deliveryProps.getDeliveryMode();
+ if(deliveryMode != null)
+ {
+ switch(deliveryMode)
+ {
+ case PERSISTENT :
+ return DeliveryMode.PERSISTENT;
+ case NON_PERSISTENT:
+ return DeliveryMode.NON_PERSISTENT;
+ default:
+ throw new JMSException("Unknown Message Delivery Mode: " + _deliveryProps.getDeliveryMode());
+ }
+ }
+ else
+ {
+ return Message.DEFAULT_DELIVERY_MODE;
+ }
+
+ }
+
+ public void setJMSDeliveryMode(int deliveryMode) throws JMSException
+ {
+ switch(deliveryMode)
+ {
+ case DeliveryMode.PERSISTENT:
+ _deliveryProps.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
+ break;
+ case DeliveryMode.NON_PERSISTENT:
+ _deliveryProps.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
+ break;
+ default:
+ throw new JMSException("Unknown JMS Delivery Mode: " + deliveryMode);
+ }
+
+ }
+
+
+ public String getJMSType() throws JMSException
+ {
+ if(getApplicationHeaders().containsKey(JMS_TYPE))
+ {
+ return getStringProperty(JMS_TYPE);
+ }
+ else
+ {
+ return null;
+ }
+ }
+
+ private Map<String, Object> getApplicationHeaders()
+ {
+ Map<String, Object> map = _messageProps.getApplicationHeaders();
+ return map == null ? Collections.EMPTY_MAP : map;
+ }
+
+ public void setJMSType(String type) throws JMSException
+ {
+ Map<String, Object> headers = _messageProps.getApplicationHeaders();
+ if(type == null)
+ {
+ if(headers != null)
+ {
+ headers.remove(JMS_TYPE);
+ }
+ }
+ else
+ {
+ if(headers == null)
+ {
+ headers = new HashMap<String,Object>();
+ _messageProps.setApplicationHeaders(headers);
+
+ }
+ headers.put(JMS_TYPE, type);
+ }
+ }
+
+ public long getJMSExpiration() throws JMSException
+ {
+ return _deliveryProps.getExpiration();
+ }
+
+ public void setJMSExpiration(long l) throws JMSException
+ {
+ _deliveryProps.setExpiration(l);
+ }
+
+
+
+ public boolean propertyExists(String propertyName) throws JMSException
+ {
+ return getApplicationHeaders().containsKey(propertyName);
+ }
+
+ public boolean getBooleanProperty(String propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+
+ Object o = getApplicationHeaders().get(propertyName);
+
+ if(o instanceof Boolean)
+ {
+ return ((Boolean)o).booleanValue();
+ }
+ else if(o instanceof String)
+ {
+ return Boolean.valueOf((String) o).booleanValue();
+ }
+ else if(getApplicationHeaders().containsKey(propertyName))
+ {
+ throw new MessageFormatException("getBooleanProperty(\""+propertyName+"\") failed as value is not boolean: " + o);
+ }
+ else
+ {
+ return Boolean.valueOf(null);
+ }
+ }
+
+ public byte getByteProperty(String propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+
+ Map<String, Object> propertyMap = getApplicationHeaders();
+
+ Object o = propertyMap.get(propertyName);
+
+ if(o instanceof Byte)
+ {
+ return ((Byte)o).byteValue();
+ }
+ else if(o instanceof String)
+ {
+ return Byte.valueOf((String) o).byteValue();
+ }
+ else if(getApplicationHeaders().containsKey(propertyName))
+ {
+ throw new MessageFormatException("getByteProperty(\""+propertyName+"\") failed as value is not a byte: " + o);
+ }
+ else
+ {
+ return Byte.valueOf(null);
+ }
+ }
+
+ public short getShortProperty(String propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+
+ Map<String, Object> propertyMap = getApplicationHeaders();
+
+ Object o = propertyMap.get(propertyName);
+
+ if(o instanceof Short)
+ {
+ return ((Short)o).shortValue();
+ }
+ else
+ {
+ try
+ {
+ return Short.valueOf(getByteProperty(propertyName));
+ }
+ catch(MessageFormatException e)
+ {
+ throw new MessageFormatException("getShortProperty(\""+propertyName+"\") failed as value is not a short: " + o);
+ }
+ }
+
+
+ }
+
+ public int getIntProperty(String propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+
+ Map<String, Object> propertyMap = getApplicationHeaders();
+
+ Object o = propertyMap.get(propertyName);
+
+ if(o instanceof Integer)
+ {
+ return ((Integer)o).intValue();
+ }
+ else
+ {
+ try
+ {
+ return Integer.valueOf(getShortProperty(propertyName));
+ }
+ catch(MessageFormatException e)
+ {
+ throw new MessageFormatException("getIntProperty(\""+propertyName+"\") failed as value is not an int: " + o);
+ }
+
+ }
+ }
+
+ public long getLongProperty(String propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+
+ Map<String, Object> propertyMap = getApplicationHeaders();
+
+ Object o = propertyMap.get(propertyName);
+
+ if(o instanceof Long)
+ {
+ return ((Long)o).longValue();
+ }
+ else
+ {
+ try
+ {
+ return Long.valueOf(getIntProperty(propertyName));
+ }
+ catch(MessageFormatException e)
+ {
+ throw new MessageFormatException("getLongProperty(\""+propertyName+"\") failed as value is not a long: " + o);
+ }
+
+ }
+ }
+
+ public float getFloatProperty(String propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ Map<String, Object> propertyMap = getApplicationHeaders();
+
+ Object o = propertyMap.get(propertyName);
+
+ if(o instanceof Float)
+ {
+ return ((Float)o).floatValue();
+ }
+ else if(o instanceof String)
+ {
+ return Float.valueOf((String) o).floatValue();
+ }
+ else if(getApplicationHeaders().containsKey(propertyName))
+ {
+ throw new MessageFormatException("getFloatProperty(\""+propertyName+"\") failed as value is not a float: " + o);
+ }
+ else
+ {
+ throw new NullPointerException("No such property: " + propertyName);
+ }
+
+ }
+
+ public double getDoubleProperty(String propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+
+ Map<String, Object> propertyMap = getApplicationHeaders();
+
+ Object o = propertyMap.get(propertyName);
+
+ if(o instanceof Double)
+ {
+ return ((Double)o).doubleValue();
+ }
+ else if (o instanceof String)
+ {
+ return Double.valueOf((String)o);
+ }
+ else
+ {
+ try
+ {
+ return Double.valueOf(getFloatProperty(propertyName));
+ }
+ catch(MessageFormatException e)
+ {
+ throw new MessageFormatException("getDoubleProperty(\""+propertyName+"\") failed as value is not a double: " + o);
+ }
+
+ }
+ }
+
+ public String getStringProperty(String propertyName) throws JMSException
+ {
+ if (propertyName.equals(CustomJMSXProperty.JMSXUserID.toString()))
+ {
+ return new String(_messageProps.getUserId());
+ }
+ else
+ {
+ checkPropertyName(propertyName);
+ Map<String, Object> propertyMap = getApplicationHeaders();
+
+ Object o = propertyMap.get(propertyName);
+
+ if(o instanceof String)
+ {
+ return (String) o;
+ }
+ else if(o == null)
+ {
+ return null;
+ }
+ else if(o.getClass().isArray())
+ {
+ throw new MessageFormatException("getString(\""+propertyName+"\") failed as value of type " + o.getClass()+ " is an array.");
+ }
+ else
+ {
+ return String.valueOf(o);
+ }
+
+ }
+ }
+
+ public Object getObjectProperty(String propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ Map<String, Object> propertyMap = getApplicationHeaders();
+
+ return propertyMap.get(propertyName);
+
+ }
+
+ public Enumeration getPropertyNames() throws JMSException
+ {
+ return java.util.Collections.enumeration(getApplicationHeaders().keySet());
+ }
+
+ public void setBooleanProperty(String propertyName, boolean b) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ checkWritableProperties();
+ setApplicationHeader(propertyName, b);
+ }
+
+ public void setByteProperty(String propertyName, byte b) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ checkWritableProperties();
+ setApplicationHeader(propertyName, b);
+ }
+
+ public void setShortProperty(String propertyName, short i) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ checkWritableProperties();
+ setApplicationHeader(propertyName, i);
+ }
+
+ public void setIntProperty(String propertyName, int i) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ checkWritableProperties();
+ setApplicationHeader(propertyName, i);
+ }
+
+ public void setLongProperty(String propertyName, long l) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ checkWritableProperties();
+ setApplicationHeader(propertyName, l);
+ }
+
+ public void setFloatProperty(String propertyName, float f) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ checkWritableProperties();
+ setApplicationHeader(propertyName, f);
+ }
+
+ public void setDoubleProperty(String propertyName, double v) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ checkWritableProperties();
+ setApplicationHeader(propertyName, v);
+ }
+
+ public void setStringProperty(String propertyName, String value) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ checkWritableProperties();
+ setApplicationHeader(propertyName, value);
+
+ if ("x-amqp-0-10.app-id".equals(propertyName)) {
+ _messageProps.setAppId(value.getBytes());
+ }
+ }
+
+ private static final Set<Class> ALLOWED = new HashSet();
+ static
+ {
+ ALLOWED.add(Boolean.class);
+ ALLOWED.add(Byte.class);
+ ALLOWED.add(Short.class);
+ ALLOWED.add(Integer.class);
+ ALLOWED.add(Long.class);
+ ALLOWED.add(Float.class);
+ ALLOWED.add(Double.class);
+ ALLOWED.add(Character.class);
+ ALLOWED.add(String.class);
+ ALLOWED.add(byte[].class);
+ }
+
+ public void setObjectProperty(String propertyName, Object object) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ checkWritableProperties();
+ if (object == null)
+ {
+ throw new MessageFormatException(AMQPInvalidClassException.INVALID_OBJECT_MSG + "null");
+ }
+ else if (!ALLOWED.contains(object.getClass()))
+ {
+ throw new MessageFormatException(AMQPInvalidClassException.INVALID_OBJECT_MSG + object.getClass());
+ }
+ setApplicationHeader(propertyName, object);
+ }
+
+ private void setApplicationHeader(String propertyName, Object object)
+ {
+ Map<String, Object> headers = _messageProps.getApplicationHeaders();
+ if(headers == null)
+ {
+ headers = new HashMap<String,Object>();
+ _messageProps.setApplicationHeaders(headers);
+ }
+ headers.put(propertyName, object);
+ }
+
+ public void removeProperty(String propertyName) throws JMSException
+ {
+ Map<String, Object> headers = _messageProps.getApplicationHeaders();
+ if(headers != null)
+ {
+ headers.remove(propertyName);
+ }
+ }
+
+
+
+ protected void checkWritableProperties() throws MessageNotWriteableException
+ {
+ if (_readableProperties)
+ {
+ throw new MessageNotWriteableException("You need to call clearProperties() to make the message writable");
+ }
+ }
+
+
+ public int getJMSPriority() throws JMSException
+ {
+ MessageDeliveryPriority messageDeliveryPriority = _deliveryProps.getPriority();
+ return messageDeliveryPriority == null ? Message.DEFAULT_PRIORITY : messageDeliveryPriority.getValue();
+ }
+
+ public void setJMSPriority(int i) throws JMSException
+ {
+ _deliveryProps.setPriority(MessageDeliveryPriority.get((short)i));
+ }
+
+ public void clearProperties() throws JMSException
+ {
+ if(!getApplicationHeaders().isEmpty())
+ {
+ getApplicationHeaders().clear();
+ }
+
+ _readableProperties = false;
+ }
+
+
+ public void acknowledgeThis() throws JMSException
+ {
+ // the JMS 1.1 spec says in section 3.6 that calls to acknowledge are ignored when client acknowledge
+ // is not specified. In our case, we only set the session field where client acknowledge mode is specified.
+ if (_session != null && _session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
+ {
+ if (_session.getAMQConnection().isClosed())
+ {
+ throw new javax.jms.IllegalStateException("Connection is already closed");
+ }
+
+ // we set multiple to true here since acknowledgment implies acknowledge of all previous messages
+ // received on the session
+ _session.acknowledgeMessage(_deliveryTag, true);
+ }
+ }
+
+ public void acknowledge() throws JMSException
+ {
+ if (_session != null && _session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
+ {
+ _session.acknowledge();
+ }
+ }
+
+
+ /**
+ * The session is set when CLIENT_ACKNOWLEDGE mode is used so that the CHANNEL ACK can be sent when the user calls
+ * acknowledge()
+ *
+ * @param s the AMQ session that delivered this message
+ */
+ public void setAMQSession(AMQSession s)
+ {
+ _session = s;
+ }
+
+ public AMQSession getAMQSession()
+ {
+ return _session;
+ }
+
+ /**
+ * Get the AMQ message number assigned to this message
+ *
+ * @return the message number
+ */
+ public long getDeliveryTag()
+ {
+ return _deliveryTag;
+ }
+
+
+
+
+
+
+ protected void checkPropertyName(CharSequence propertyName)
+ {
+ if (propertyName == null)
+ {
+ throw new IllegalArgumentException("Property name must not be null");
+ }
+ else if (propertyName.length() == 0)
+ {
+ throw new IllegalArgumentException("Property name must not be the empty string");
+ }
+
+ checkIdentiferFormat(propertyName);
+ }
+
+ protected void checkIdentiferFormat(CharSequence propertyName)
+ {
+// JMS requirements 3.5.1 Property Names
+// Identifiers:
+// - An identifier is an unlimited-length character sequence that must begin
+// with a Java identifier start character; all following characters must be Java
+// identifier part characters. An identifier start character is any character for
+// which the method Character.isJavaIdentifierStart returns true. This includes
+// '_' and '$'. An identifier part character is any character for which the
+// method Character.isJavaIdentifierPart returns true.
+// - Identifiers cannot be the names NULL, TRUE, or FALSE.
+// Identifiers cannot be NOT, AND, OR, BETWEEN, LIKE, IN, IS, or
+// ESCAPE.
+// Identifiers are either header field references or property references. The
+// type of a property value in a message selector corresponds to the type
+// used to set the property. If a property that does not exist in a message is
+// referenced, its value is NULL. The semantics of evaluating NULL values
+// in a selector are described in Section 3.8.1.2, Null Values.
+// The conversions that apply to the get methods for properties do not
+// apply when a property is used in a message selector expression. For
+// example, suppose you set a property as a string value, as in the
+// following:
+// myMessage.setStringProperty("NumberOfOrders", "2");
+// The following expression in a message selector would evaluate to false,
+// because a string cannot be used in an arithmetic expression:
+// "NumberOfOrders > 1"
+// Identifiers are case sensitive.
+// Message header field references are restricted to JMSDeliveryMode,
+// JMSPriority, JMSMessageID, JMSTimestamp, JMSCorrelationID, and
+// JMSType. JMSMessageID, JMSCorrelationID, and JMSType values may be
+// null and if so are treated as a NULL value.
+
+ if (Boolean.getBoolean("strict-jms"))
+ {
+ // JMS start character
+ if (!(Character.isJavaIdentifierStart(propertyName.charAt(0))))
+ {
+ throw new IllegalArgumentException("Identifier '" + propertyName + "' does not start with a valid JMS identifier start character");
+ }
+
+ // JMS part character
+ int length = propertyName.length();
+ for (int c = 1; c < length; c++)
+ {
+ if (!(Character.isJavaIdentifierPart(propertyName.charAt(c))))
+ {
+ throw new IllegalArgumentException("Identifier '" + propertyName + "' contains an invalid JMS identifier character");
+ }
+ }
+
+ // JMS invalid names
+ if ((propertyName.equals("NULL")
+ || propertyName.equals("TRUE")
+ || propertyName.equals("FALSE")
+ || propertyName.equals("NOT")
+ || propertyName.equals("AND")
+ || propertyName.equals("OR")
+ || propertyName.equals("BETWEEN")
+ || propertyName.equals("LIKE")
+ || propertyName.equals("IN")
+ || propertyName.equals("IS")
+ || propertyName.equals("ESCAPE")))
+ {
+ throw new IllegalArgumentException("Identifier '" + propertyName + "' is not allowed in JMS");
+ }
+ }
+
+ }
+
+
+ public MessageProperties getMessageProperties()
+ {
+ return _messageProps;
+ }
+
+
+ public DeliveryProperties getDeliveryProperties()
+ {
+ return _deliveryProps;
+ }
+}
Propchange: qpid/trunk/qpid/tools/src/java/src/patch/java/AMQMessageDelegate_0_10.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/tools/src/java/src/patch/java/AMQMessageDelegate_0_10.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org