From 1ee9cd5f7d43bd8c8117eab317842dd3783fac63 Mon Sep 17 00:00:00 2001
From: Robert Godfrey <rgodfrey@apache.org>
Date: Thu, 20 Feb 2014 15:50:09 +0000
Subject: QPID-5563 : Ensure vhost is set as early as possible, on receipt of
 connection open

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1570242 13f79535-47bb-0310-9956-ffa450edef68
---
 .../amqp_1_0/transport/ConnectionEndpoint.java     |  2 +
 .../transport/ConnectionEventListener.java         |  8 ++++
 .../qpid/server/protocol/v1_0/Connection_1_0.java  | 54 +++++++++++-----------
 .../protocol/v1_0/ProtocolEngine_1_0_0_SASL.java   |  1 -
 .../qpid/server/protocol/v1_0/Session_1_0.java     | 42 +++++++++++------
 5 files changed, 65 insertions(+), 42 deletions(-)

(limited to 'qpid/java')

diff --git a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
index 77880fb145..4e1a90d4d2 100644
--- a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
+++ b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
@@ -343,6 +343,8 @@ public class ConnectionEndpoint implements DescribedTypeConstructorRegistry.Sour
             _idleTimeout = open.getIdleTimeOut().longValue();
         }
 
+        _connectionEventListener.openReceived();
+
         switch (_state)
         {
             case UNOPENED:
diff --git a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEventListener.java b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEventListener.java
index 17163598d2..ba7ff86e30 100644
--- a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEventListener.java
+++ b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEventListener.java
@@ -21,8 +21,16 @@ package org.apache.qpid.amqp_1_0.transport;
 
 public interface ConnectionEventListener
 {
+    void openReceived();
+
     class DefaultConnectionEventListener implements ConnectionEventListener
     {
+        @Override
+        public void openReceived()
+        {
+
+        }
+
         public void remoteSessionCreation(final SessionEndpoint endpoint)
         {
             endpoint.end();
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
index 5e38442d7e..0a53a6436a 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
@@ -20,7 +20,6 @@
  */
 package org.apache.qpid.server.protocol.v1_0;
 
-import java.net.InetAddress;
 import java.net.SocketAddress;
 import java.security.Principal;
 import java.security.PrivilegedAction;
@@ -121,9 +120,32 @@ public class Connection_1_0 implements ConnectionEventListener, AMQConnectionMod
         return _reference;
     }
 
+    @Override
+    public void openReceived()
+    {
+        String host = _conn.getLocalHostname();
+        if(host == null || host.trim().equals(""))
+        {
+            host = (String)_broker.getAttribute(Broker.DEFAULT_VIRTUAL_HOST);
+        }
+        _vhost = _broker.getVirtualHostRegistry().getVirtualHost(host);
+        if(_vhost == null)
+        {
+            final Error err = new Error();
+            err.setCondition(AmqpError.NOT_FOUND);
+            err.setDescription("Unknown hostname " + _conn.getLocalHostname());
+            _conn.close(err);
+        }
+        Subject authSubject = _subjectCreator.createSubjectWithGroups(_conn.getUser());
+        _subject.getPrincipals().addAll(authSubject.getPrincipals());
+        _subject.getPublicCredentials().addAll(authSubject.getPublicCredentials());
+        _subject.getPrivateCredentials().addAll(authSubject.getPrivateCredentials());
+
+    }
+
     public void remoteSessionCreation(SessionEndpoint endpoint)
     {
-        final Session_1_0 session = new Session_1_0(_vhost, this, endpoint);
+        final Session_1_0 session = new Session_1_0(this, endpoint);
         _sessions.add(session);
         endpoint.setSessionEventListener(new SessionEventListener()
         {
@@ -397,33 +419,13 @@ public class Connection_1_0 implements ConnectionEventListener, AMQConnectionMod
     }
 
 
-    public void frameReceived()
+    Subject getSubject()
     {
-        if(_vhost == null && _conn.isOpen())
-        {
-            String host = _conn.getLocalHostname();
-            if(host == null || host.trim().equals(""))
-            {
-                host = (String)_broker.getAttribute(Broker.DEFAULT_VIRTUAL_HOST);
-            }
-            _vhost = _broker.getVirtualHostRegistry().getVirtualHost(host);
-            if(_vhost == null)
-            {
-                final Error err = new Error();
-                err.setCondition(AmqpError.NOT_FOUND);
-                err.setDescription("Unknown hostname " + _conn.getLocalHostname());
-                _conn.close(err);
-            }
-            Subject authSubject = _subjectCreator.createSubjectWithGroups(_conn.getUser());
-            _subject.getPrincipals().addAll(authSubject.getPrincipals());
-            _subject.getPublicCredentials().addAll(authSubject.getPublicCredentials());
-            _subject.getPrivateCredentials().addAll(authSubject.getPrivateCredentials());
-
-        }
+        return _subject;
     }
 
-    Subject getSubject()
+    VirtualHost getVirtualHost()
     {
-        return _subject;
+        return _vhost;
     }
 }
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
index 468ba2758e..f82689e36a 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
@@ -365,7 +365,6 @@ public class ProtocolEngine_1_0_0_SASL implements ServerProtocolEngine, FrameOut
                             public Void run()
                             {
                                 _frameHandler = _frameHandler.parse(msg);
-                                _connection.frameReceived();
                                 return null;
                             }
                         });
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index dcf9e5d7a1..ac1517aaf5 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -87,11 +87,9 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
 
 
 
-    public Session_1_0(VirtualHost vhost, final Connection_1_0 connection, final SessionEndpoint endpoint)
+    public Session_1_0(final Connection_1_0 connection, final SessionEndpoint endpoint)
     {
-        _vhost = vhost;
         _endpoint = endpoint;
-        _transaction = new AutoCommitTransaction(vhost.getMessageStore());
         _connection = connection;
         _subject.getPrincipals().addAll(connection.getSubject().getPrincipals());
         _subject.getPrincipals().add(new SessionPrincipal(this));
@@ -106,7 +104,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
 
         final
         LinkRegistry
-                linkRegistry = _vhost.getLinkRegistry(endpoint.getSession().getConnection().getRemoteContainerId());
+                linkRegistry = getVirtualHost().getLinkRegistry(endpoint.getSession().getConnection().getRemoteContainerId());
 
 
         if(endpoint.getRole() == Role.SENDER)
@@ -129,7 +127,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
                         source.setAddress(tempQueue.getName());
                     }
                     String addr = source.getAddress();
-                    MessageSource queue = _vhost.getMessageSource(addr);
+                    MessageSource queue = getVirtualHost().getMessageSource(addr);
                     if(queue != null)
                     {
 
@@ -140,7 +138,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
                     }
                     else
                     {
-                        Exchange exchg = _vhost.getExchange(addr);
+                        Exchange exchg = getVirtualHost().getExchange(addr);
                         if(exchg != null)
                         {
                             destination = new ExchangeDestination(exchg, source.getDurable(), source.getExpiryPolicy());
@@ -165,7 +163,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
                     try
                     {
                         final SendingLink_1_0 sendingLink = new SendingLink_1_0(new SendingLinkAttachment(this, sendingLinkEndpoint),
-                                                                                _vhost,
+                                                                                getVirtualHost(),
                                                                                 (SendingDestination) destination
                         );
 
@@ -245,7 +243,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
 
                 final ReceivingLinkEndpoint receivingLinkEndpoint = (ReceivingLinkEndpoint) endpoint;
                 final TxnCoordinatorLink_1_0 coordinatorLink =
-                        new TxnCoordinatorLink_1_0(_vhost, this, receivingLinkEndpoint, _openTransactions);
+                        new TxnCoordinatorLink_1_0(getVirtualHost(), this, receivingLinkEndpoint, _openTransactions);
                 receivingLinkEndpoint.setLinkEventListener(new SubjectSpecificReceivingLinkListener(coordinatorLink));
                 link = coordinatorLink;
 
@@ -272,7 +270,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
                         }
 
                         String addr = target.getAddress();
-                        MessageDestination messageDestination = _vhost.getMessageDestination(addr);
+                        MessageDestination messageDestination = getVirtualHost().getMessageDestination(addr);
                         if(messageDestination != null)
                         {
                             destination = new NodeReceivingDestination(messageDestination, target.getDurable(),
@@ -280,7 +278,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
                         }
                         else
                         {
-                            AMQQueue queue = _vhost.getQueue(addr);
+                            AMQQueue queue = getVirtualHost().getQueue(addr);
                             if(queue != null)
                             {
 
@@ -303,7 +301,8 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
                     if(destination != null)
                     {
                         final ReceivingLinkEndpoint receivingLinkEndpoint = (ReceivingLinkEndpoint) endpoint;
-                        final ReceivingLink_1_0 receivingLink = new ReceivingLink_1_0(new ReceivingLinkAttachment(this, receivingLinkEndpoint), _vhost,
+                        final ReceivingLink_1_0 receivingLink = new ReceivingLink_1_0(new ReceivingLinkAttachment(this, receivingLinkEndpoint),
+                                                                                      getVirtualHost(),
                                 (ReceivingDestination) destination);
 
                         receivingLinkEndpoint.setLinkEventListener(new SubjectSpecificReceivingLinkListener(receivingLink));
@@ -355,7 +354,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
                                             ? null
                                             : (LifetimePolicy) properties.get(LIFETIME_POLICY);
             Map<String,Object> attributes = new HashMap<String,Object>();
-            attributes.put(org.apache.qpid.server.model.Queue.ID, UUIDGenerator.generateQueueUUID(queueName, _vhost.getName()));
+            attributes.put(org.apache.qpid.server.model.Queue.ID, UUIDGenerator.generateQueueUUID(queueName, getVirtualHost().getName()));
             attributes.put(org.apache.qpid.server.model.Queue.NAME, queueName);
             attributes.put(org.apache.qpid.server.model.Queue.DURABLE, false);
 
@@ -388,7 +387,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
 
             // TODO convert AMQP 1-0 node properties to queue attributes
 
-            final AMQQueue tempQueue = queue = _vhost.createQueue(attributes );
+            final AMQQueue tempQueue = queue = getVirtualHost().createQueue(attributes);
         }
         catch (AccessControlException e)
         {
@@ -409,7 +408,15 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
     {
         // TODO should treat invalid id differently to null
         ServerTransaction transaction = _openTransactions.get(binaryToInteger(transactionId));
-        return transaction == null ? _transaction : transaction;
+        if(transaction == null)
+        {
+            if(_transaction == null)
+            {
+                _transaction = new AutoCommitTransaction(_connection.getVirtualHost().getMessageStore());
+            }
+            transaction = _transaction;
+        }
+        return transaction;
     }
 
     public void remoteEnd(End end)
@@ -622,7 +629,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
                                     connectionId,
                                     getClientID(),
                                     remoteAddress,
-                                    _vhost.getName(),
+                                    getVirtualHost().getName(),
                                     _endpoint.getSendingChannel())  + "] ";
     }
 
@@ -657,6 +664,11 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
         return _subject;
     }
 
+    VirtualHost getVirtualHost()
+    {
+        return _connection.getVirtualHost();
+    }
+
 
     private class SubjectSpecificReceivingLinkListener implements ReceivingLinkListener
     {
-- 
cgit v1.2.1