You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2011/08/29 00:34:40 UTC
svn commit: r1162604 - in /qpid/branches/rg-amqp-1-0-sandbox/qpid/java:
amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/
amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/
amqp-1-0-client/src/main/java/o...
Author: rgodfrey
Date: Sun Aug 28 22:34:40 2011
New Revision: 1162604
URL: http://svn.apache.org/viewvc?rev=1162604&view=rev
Log:
NO-JIRA: AMQP 1-0 Sandbox - fix multiple sessions, add preliminary onMessage support, example
Added:
qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/Hello.java
qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/hello.properties
Modified:
qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java
qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/SessionImpl.java
qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java
qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
Added: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/Hello.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/Hello.java?rev=1162604&view=auto
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/Hello.java (added)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/Hello.java Sun Aug 28 22:34:40 2011
@@ -0,0 +1,141 @@
+/*
+ *
+ * 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.amqp_1_0.jms.example;
+
+import javax.jms.*;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.Hashtable;
+import java.util.Map;
+import java.util.Properties;
+
+
+public class Hello
+{
+
+ public Hello()
+ {
+ }
+
+ public static void main(String[] args)
+ {
+ try
+ {
+ Class.forName("org.apache.qpid.amqp_1_0.jms.jndi.PropertiesFileInitialContextFactory");
+
+ Hashtable env = new Hashtable();
+ env.put("java.naming.provider.url", "hello.properties");
+ env.put("java.naming.factory.initial", "org.apache.qpid.amqp_1_0.jms.jndi.PropertiesFileInitialContextFactory");
+
+ Context context = new InitialContext(env);
+
+ ConnectionFactory connectionFactory = (ConnectionFactory) context.lookup("localhost");
+ Connection connection = connectionFactory.createConnection();
+
+ Session producersession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = (Queue) context.lookup("queue");
+
+
+ Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageConsumer messageConsumer = consumerSession.createConsumer(queue);
+
+ messageConsumer.setMessageListener(new MessageListener()
+ {
+ public void onMessage(final Message message)
+ {
+ try
+ {
+
+ if(message instanceof TextMessage)
+ {
+ System.out.println("Received text Message:");
+ System.out.println("======================");
+ System.out.println(((TextMessage) message).getText());
+ }
+ else if(message instanceof MapMessage)
+ {
+ System.out.println("Received Map Message:");
+ System.out.println("=====================");
+
+
+ MapMessage mapmessage = (MapMessage) message;
+
+ Enumeration names = mapmessage.getMapNames();
+
+ while(names.hasMoreElements())
+ {
+ String name = (String) names.nextElement();
+ System.out.println(name + " -> " + mapmessage.getObject(name));
+ }
+
+ }
+ else if(message instanceof BytesMessage)
+ {
+ System.out.println("Received Bytes Message:");
+ System.out.println("=======================");
+ System.out.println(((BytesMessage) message).readUTF());
+ }
+ }
+ catch (JMSException e)
+ {
+ e.printStackTrace(); //TODO
+ }
+
+ }
+ });
+
+ connection.start();
+
+
+ MessageProducer messageProducer = producersession.createProducer(queue);
+ TextMessage message = producersession.createTextMessage("Hello world!");
+ messageProducer.send(message);
+
+ MapMessage mapmessage = producersession.createMapMessage();
+ mapmessage.setBoolean("mybool", true);
+ mapmessage.setString("mystring", "hello");
+ mapmessage.setLong("mylong", -25L);
+
+
+ messageProducer.send(mapmessage);
+
+ BytesMessage bytesMessage = producersession.createBytesMessage();
+ bytesMessage.writeUTF("This is a bytes message");
+
+ messageProducer.send(bytesMessage);
+
+
+
+ Thread.sleep(50000L);
+
+ connection.close();
+ context.close();
+ }
+ catch (Exception exp)
+ {
+ exp.printStackTrace();
+ }
+ }
+}
Added: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/hello.properties
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/hello.properties?rev=1162604&view=auto
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/hello.properties (added)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/example/src/main/java/org/apache/qpid/amqp_1_0/jms/example/hello.properties Sun Aug 28 22:34:40 2011
@@ -0,0 +1,28 @@
+#
+# 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.
+#
+java.naming.factory.initial = org.apache.qpid.amqp_1_0.jms.jndi.PropertiesFileInitialContextFactory
+
+# register some connection factories
+# connectionfactory.[jndiname] = [ConnectionURL]
+connectionfactory.localhost = http://guest:guest@localhost/test?cliendId='test-client'
+
+
+# Register an AMQP destination in JNDI
+# destination.[jniName] = [Address Format]
+queue.queue = queue
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java?rev=1162604&r1=1162603&r2=1162604&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java Sun Aug 28 22:34:40 2011
@@ -18,6 +18,7 @@
*/
package org.apache.qpid.amqp_1_0.jms.impl;
+import org.apache.qpid.amqp_1_0.client.Message;
import org.apache.qpid.amqp_1_0.client.Receiver;
import org.apache.qpid.amqp_1_0.jms.MessageConsumer;
import org.apache.qpid.amqp_1_0.type.Binary;
@@ -36,6 +37,7 @@ public class MessageConsumerImpl impleme
private SessionImpl _session;
private Receiver _receiver;
private Binary _lastUnackedMessage;
+ private MessageListener _messageListener;
MessageConsumerImpl(final Destination destination,
final SessionImpl session,
@@ -68,14 +70,23 @@ public class MessageConsumerImpl impleme
return _selector;
}
- public MessageListener getMessageListener() throws JMSException
+ public MessageListener getMessageListener()
{
- return null; //TODO
+ return _messageListener;
}
public void setMessageListener(final MessageListener messageListener) throws JMSException
{
- //TODO
+ _messageListener = messageListener;
+ _session.messageListenerSet( this );
+ _receiver.setMessageArrivalListener(new Receiver.MessageArrivalListener()
+ {
+
+ public void messageArrived(final Receiver receiver)
+ {
+ _session.messageArrived(MessageConsumerImpl.this);
+ }
+ });
}
public MessageImpl receive() throws JMSException
@@ -97,7 +108,7 @@ public class MessageConsumerImpl impleme
private MessageImpl receiveImpl(long timeout)
{
- org.apache.qpid.amqp_1_0.client.Message msg = _receiver.receive(timeout);
+ org.apache.qpid.amqp_1_0.client.Message msg = receive0(timeout);
if(msg != null)
{
preReceiveAction(msg);
@@ -105,13 +116,18 @@ public class MessageConsumerImpl impleme
return createJMSMessage(msg);
}
+ Message receive0(final long timeout)
+ {
+ return _receiver.receive(timeout);
+ }
+
void acknowledge(final org.apache.qpid.amqp_1_0.client.Message msg)
{
_receiver.acknowledge(msg.getDeliveryTag());
}
- private MessageImpl createJMSMessage(final org.apache.qpid.amqp_1_0.client.Message msg)
+ MessageImpl createJMSMessage(final org.apache.qpid.amqp_1_0.client.Message msg)
{
if(msg != null)
{
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/SessionImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/SessionImpl.java?rev=1162604&r1=1162603&r2=1162604&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/SessionImpl.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/SessionImpl.java Sun Aug 28 22:34:40 2011
@@ -19,6 +19,7 @@
package org.apache.qpid.amqp_1_0.jms.impl;
import org.apache.qpid.amqp_1_0.client.Connection;
+import org.apache.qpid.amqp_1_0.client.Message;
import org.apache.qpid.amqp_1_0.jms.Session;
import javax.jms.Destination;
@@ -37,6 +38,10 @@ public class SessionImpl implements Sess
private org.apache.qpid.amqp_1_0.client.Session _session;
private MessageFactory _messageFactory;
private List<MessageConsumerImpl> _consumers = new ArrayList<MessageConsumerImpl>();
+ private MessageListener _messageListener;
+ private Dispatcher _dispatcher = new Dispatcher();
+ private Thread _dispatcherThread;
+
protected SessionImpl(final ConnectionImpl connection, final AcknowledgeMode acknowledgeMode)
{
@@ -45,6 +50,9 @@ public class SessionImpl implements Sess
Connection clientConn = _connection.getClientConnection();
_session = clientConn.createSession();
_messageFactory = new MessageFactory(this);
+
+ _dispatcherThread = new Thread(_dispatcher);
+ _dispatcherThread.start();
}
public BytesMessageImpl createBytesMessage() throws JMSException
@@ -119,7 +127,9 @@ public class SessionImpl implements Sess
public void close() throws JMSException
{
+ _dispatcher.close();
_session.close();
+
}
public void recover() throws JMSException
@@ -129,12 +139,19 @@ public class SessionImpl implements Sess
public MessageListener getMessageListener() throws JMSException
{
- return null; //TODO
+ return _messageListener;
}
public void setMessageListener(final MessageListener messageListener) throws JMSException
{
- //TODO
+ if(_messageListener != null)
+ {
+
+ }
+ else
+ {
+ _messageListener = messageListener;
+ }
}
public void run()
@@ -231,7 +248,11 @@ public class SessionImpl implements Sess
void start()
{
- //TODO
+ _dispatcher.start();
+ for(MessageConsumerImpl consumer : _consumers)
+ {
+ consumer.start();
+ }
}
org.apache.qpid.amqp_1_0.client.Session getClientSession()
@@ -254,4 +275,116 @@ public class SessionImpl implements Sess
}
}
}
+
+ void messageListenerSet(final MessageConsumerImpl messageConsumer)
+ {
+ _dispatcher.updateMessageListener(messageConsumer);
+ }
+
+ public void messageArrived(final MessageConsumerImpl messageConsumer)
+ {
+ _dispatcher.messageArrivedAtConsumer(messageConsumer);
+ }
+
+
+ private class Dispatcher implements Runnable
+ {
+
+ private final List<MessageConsumerImpl> _messageConsumerList = new ArrayList<MessageConsumerImpl>();
+
+ private boolean _closed;
+ private boolean _started;
+
+ public void run()
+ {
+ synchronized(getLock())
+ {
+ while(!_closed)
+ {
+ while(!_started || _messageConsumerList.isEmpty())
+ {
+ try
+ {
+ getLock().wait();
+ }
+ catch (InterruptedException e)
+ {
+ return;
+ }
+ }
+ while(_started && !_messageConsumerList.isEmpty())
+ {
+ MessageConsumerImpl consumer = _messageConsumerList.remove(0);
+ MessageListener listener = consumer.getMessageListener();
+ Message msg = consumer.receive0(0L);
+
+ MessageImpl message = consumer.createJMSMessage(msg);
+
+ if(message != null)
+ {
+ listener.onMessage(message);
+ if(_acknowledgeMode == AcknowledgeMode.AUTO_ACKNOWLEDGE
+ || _acknowledgeMode == AcknowledgeMode.DUPS_OK_ACKNOWLEDGE)
+ {
+ consumer.acknowledge(msg);
+ }
+ }
+
+ }
+ }
+
+
+ }
+
+ }
+
+ private Object getLock()
+ {
+ return _session.getEndpoint().getLock();
+ }
+
+ public void messageArrivedAtConsumer(MessageConsumerImpl impl)
+ {
+ synchronized (getLock())
+ {
+ _messageConsumerList.add(impl);
+ getLock().notifyAll();
+ }
+ }
+
+ public void close()
+ {
+ synchronized (getLock())
+ {
+ _closed = true;
+ getLock().notifyAll();
+ }
+ }
+
+ public void updateMessageListener(final MessageConsumerImpl messageConsumer)
+ {
+ synchronized (getLock())
+ {
+ getLock().notifyAll();
+ }
+ }
+
+ public void start()
+ {
+ synchronized (getLock())
+ {
+ _started = true;
+ getLock().notifyAll();
+ }
+ }
+
+ public void stop()
+ {
+ synchronized (getLock())
+ {
+ _started = false;
+ getLock().notifyAll();
+ }
+ }
+ }
}
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java?rev=1162604&r1=1162603&r2=1162604&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java Sun Aug 28 22:34:40 2011
@@ -48,6 +48,7 @@ public class Receiver implements Deliver
private Queue<Transfer> _prefetchQueue = new ConcurrentLinkedQueue<Transfer>();
private Map<Binary, SettledAction> _unsettledMap = new HashMap<Binary, SettledAction>();
+ private MessageArrivalListener _messageArrivalListener;
public Receiver(final Session session,
final String linkName,
@@ -110,6 +111,7 @@ public class Receiver implements Deliver
@Override public void messageTransfer(final Transfer xfr)
{
_prefetchQueue.add(xfr);
+ postPrefetchAction();
}
});
@@ -133,6 +135,14 @@ public class Receiver implements Deliver
}
}
+ private void postPrefetchAction()
+ {
+ if(_messageArrivalListener != null)
+ {
+ _messageArrivalListener.messageArrived(this);
+ }
+ }
+
public void setCredit(UnsignedInteger credit, boolean window)
{
_endpoint.setLinkCredit(credit);
@@ -262,7 +272,7 @@ public class Receiver implements Deliver
}
catch (InterruptedException e)
{
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ return null;
}
if(wait > 0L)
{
@@ -393,8 +403,22 @@ public class Receiver implements Deliver
}
+ public void setMessageArrivalListener(final MessageArrivalListener messageArrivalListener)
+ {
+ synchronized(_endpoint.getLock())
+ {
+ _messageArrivalListener = messageArrivalListener;
+ }
+ }
+
public static interface SettledAction
{
public void onSettled(Binary deliveryTag);
}
+
+
+ public interface MessageArrivalListener
+ {
+ void messageArrived(Receiver receiver);
+ }
}
\ No newline at end of file
Modified: qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
URL: http://svn.apache.org/viewvc/qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java?rev=1162604&r1=1162603&r2=1162604&view=diff
==============================================================================
--- qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java (original)
+++ qpid/branches/rg-amqp-1-0-sandbox/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java Sun Aug 28 22:34:40 2011
@@ -439,6 +439,7 @@ public class ConnectionEndpoint implemen
SessionEndpoint endpoint = new SessionEndpoint(this,begin);
_receivingSessions[channel] = endpoint;
+ _sendingSessions[myChannelId] = endpoint;
Begin beginToSend = new Begin();
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org