Author: rgreig
Date: Tue Dec 19 01:29:19 2006
New Revision: 488596
URL: http://svn.apache.org/viewvc?view=rev&rev=488596
Log:
QPID-215 : Patch supplied by Rob Godfrey - Implement custom JMSX properties
Added:
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java
Modified:
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
Modified:
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java?view=diff&rev=488596&r1=488595&r2=488596
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
(original)
+++
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
Tue Dec 19 01:29:19 2006
@@ -139,6 +139,12 @@
*/
private AMQException _lastAMQException = null;
+
+ /*
+ * The connection meta data
+ */
+ private QpidConnectionMetaData _connectionMetaData;
+
public AMQConnection(String broker, String username, String password,
String clientName, String virtualHost) throws
AMQException, URLSyntaxException
{
@@ -281,6 +287,7 @@
throw e;
}
+ _connectionMetaData = new QpidConnectionMetaData(this);
}
protected boolean checkException(Throwable thrown)
@@ -550,7 +557,7 @@
public ConnectionMetaData getMetaData() throws JMSException
{
checkNotClosed();
- return QpidConnectionMetaData.instance();
+ return _connectionMetaData;
}
Modified:
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?view=diff&rev=488596&r1=488595&r2=488596
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
(original)
+++
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
Tue Dec 19 01:29:19 2006
@@ -22,6 +22,8 @@
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
+import org.apache.qpid.url.AMQBindingURL;
+import org.apache.qpid.url.URLSyntaxException;
import org.apache.qpid.client.message.AbstractJMSMessage;
import org.apache.qpid.client.message.MessageFactoryRegistry;
import org.apache.qpid.client.message.UnprocessedMessage;
@@ -39,6 +41,7 @@
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
+import javax.jms.Destination;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -241,6 +244,17 @@
if(_session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
{
_unacknowledgedDeliveryTags.add(jmsMsg.getDeliveryTag());
+ String url =
jmsMsg.getStringProperty(CustomJMXProperty.JMSX_QPID_JMSDESTINATIONURL.toString());
+ try
+ {
+ Destination dest = AMQDestination.createDestination(new
AMQBindingURL(url));
+ jmsMsg.setJMSDestination(dest);
+ }
+ catch (URLSyntaxException e)
+ {
+ _logger.warn("Unable to parse the supplied destination header:
" + url);
+ }
+
}
_session.setInRecovery(false);
}
Modified:
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java?view=diff&rev=488596&r1=488595&r2=488596
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
(original)
+++
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
Tue Dec 19 01:29:19 2006
@@ -507,8 +507,11 @@
long timeToLive, boolean mandatory, boolean
immediate, boolean wait) throws JMSException
{
checkTemporaryDestination(destination);
+ origMessage.setJMSDestination(destination);
+
AbstractJMSMessage message = convertToNativeMessage(origMessage);
+
message.getJmsContentHeaderProperties().getJMSHeaders().setString(CustomJMXProperty.JMSX_QPID_JMSDESTINATIONURL.toString(),
destination.toURL());
AMQFrame publishFrame = BasicPublishBody.createAMQFrame(_channelId, 0,
destination.getExchangeName(),
destination.getRoutingKey(), mandatory, immediate);
Added:
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java?view=auto&rev=488596
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java
(added)
+++
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/CustomJMXProperty.java
Tue Dec 19 01:29:19 2006
@@ -0,0 +1,47 @@
+/*
+ *
+ * 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;
+
+import java.util.*;
+
+public enum CustomJMXProperty
+{
+ JMSX_QPID_JMSDESTINATIONURL,
+ JMSXGroupID,
+ JMSXGroupSeq;
+
+ private static Enumeration _names;
+
+ public static synchronized Enumeration asEnumeration()
+ {
+ if(_names == null)
+ {
+ CustomJMXProperty[] properties = values();
+ ArrayList<String> nameList = new
ArrayList<String>(properties.length);
+ for(CustomJMXProperty property : properties)
+ {
+ nameList.add(property.toString());
+ }
+ _names = Collections.enumeration(nameList);
+ }
+ return _names;
+ }
+}
Modified:
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java?view=diff&rev=488596&r1=488595&r2=488596
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java
(original)
+++
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/QpidConnectionMetaData.java
Tue Dec 19 01:29:19 2006
@@ -1,3 +1,23 @@
+/*
+ *
+ * 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;
import java.util.Enumeration;
@@ -5,46 +25,73 @@
import javax.jms.ConnectionMetaData;
import javax.jms.JMSException;
-public class QpidConnectionMetaData implements ConnectionMetaData {
+public class QpidConnectionMetaData implements ConnectionMetaData
+{
+
+
+
+ QpidConnectionMetaData(AMQConnection conn)
+ {
+ }
+
+ public int getJMSMajorVersion() throws JMSException
+ {
+ return 1;
+ }
+
+ public int getJMSMinorVersion() throws JMSException
+ {
+ return 1;
+ }
+
+ public String getJMSProviderName() throws JMSException
+ {
+ return "Apache Qpid";
+ }
+
+ public String getJMSVersion() throws JMSException
+ {
+ return "1.1";
+ }
+
+ public Enumeration getJMSXPropertyNames() throws JMSException
+ {
+ return CustomJMXProperty.asEnumeration();
+ }
+
+ public int getProviderMajorVersion() throws JMSException
+ {
+ return 0;
+ }
+
+ public int getProviderMinorVersion() throws JMSException
+ {
+ return 8;
+ }
+
+ public String getProviderVersion() throws JMSException
+ {
+ return "QPID (Client: [" + getClientVersion() + "] ; Broker [" +
getBrokerVersion() + "] ; Protocol: [ "
+ + getProtocolVersion() + "] )";
+ }
+
+ private String getProtocolVersion()
+ {
+ // TODO - Implement based on connection negotiated protocol
+ return "0.8";
+ }
+
+ public String getBrokerVersion()
+ {
+ // TODO - get broker version
+ return "<unkown>";
+ }
+
+ public String getClientVersion()
+ {
+ // TODO - get client build version from properties file or similar
+ return "<unknown>";
+ }
+
- private static QpidConnectionMetaData _instance = new
QpidConnectionMetaData();
-
- private QpidConnectionMetaData(){
- }
-
- public static QpidConnectionMetaData instance(){
- return _instance;
- }
-
- public int getJMSMajorVersion() throws JMSException {
- return 1;
- }
-
- public int getJMSMinorVersion() throws JMSException {
- return 1;
- }
-
- public String getJMSProviderName() throws JMSException {
- return "Apache Qpid";
- }
-
- public String getJMSVersion() throws JMSException {
- return "1.1";
- }
-
- public Enumeration getJMSXPropertyNames() throws JMSException {
- return null;
- }
-
- public int getProviderMajorVersion() throws JMSException {
- return 0;
- }
-
- public int getProviderMinorVersion() throws JMSException {
- return 9;
- }
-
- public String getProviderVersion() throws JMSException {
- return "Incubating-M1";
- }
}
Modified:
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java?view=diff&rev=488596&r1=488595&r2=488596
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
(original)
+++
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
Tue Dec 19 01:29:19 2006
@@ -172,13 +172,12 @@
public Destination getJMSDestination() throws JMSException
{
- // TODO: implement this once we have sorted out how to figure out the
exchange class
- return _destination;
+ return _destination;
}
public void setJMSDestination(Destination destination) throws JMSException
{
- _destination = destination;
+ _destination = destination;
}
public int getJMSDeliveryMode() throws JMSException
Added:
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java?view=auto&rev=488596
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java
(added)
+++
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java
Tue Dec 19 01:29:19 2006
@@ -0,0 +1,70 @@
+package org.apache.qpid.test.unit.message;
+
+import junit.framework.TestCase;
+import org.apache.log4j.Logger;
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.AMQHeadersExchange;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.url.AMQBindingURL;
+import org.apache.qpid.url.BindingURL;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.PropertyFieldTable;
+
+import javax.jms.*;
+
+/**
+ * @author Apache Software Foundation
+ */
+public class JMSDestinationTest extends TestCase
+{
+
+ private static final Logger _logger =
Logger.getLogger(JMSDestinationTest.class);
+
+ public String _connectionString = "vm://:1";
+
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+ TransportConnection.createVMBroker(1);
+ }
+
+
+ protected void tearDown() throws Exception
+ {
+ super.tearDown();
+ TransportConnection.killAllVMBrokers();
+ }
+
+ public void testJMSDestination() throws Exception
+ {
+ Connection con = new AMQConnection("vm://:1", "guest", "guest",
"consumer1", "/test");
+ AMQSession consumerSession = (AMQSession) con.createSession(false,
Session.CLIENT_ACKNOWLEDGE);
+ Queue queue = new AMQQueue("someQ", "someQ", false, true);
+ MessageConsumer consumer = consumerSession.createConsumer(queue);
+
+ Connection con2 = new AMQConnection("vm://:1", "guest", "guest",
"producer1", "/test");
+ Session producerSession = con2.createSession(false,
Session.CLIENT_ACKNOWLEDGE);
+ MessageProducer producer = producerSession.createProducer(queue);
+
+ TextMessage sentMsg = producerSession.createTextMessage("hello");
+ assertNull(sentMsg.getJMSDestination());
+
+ producer.send(sentMsg);
+
+ assertEquals(sentMsg.getJMSDestination(), queue);
+
+ con2.close();
+
+ con.start();
+
+ TextMessage rm = (TextMessage) consumer.receive();
+ assertNotNull(rm);
+
+ assertEquals(rm.getJMSDestination(),queue);
+ con.close();
+ }
+
+}
