QPID-6029 : [Java Broker] ConfiguredObjectRecordConverter should allow reference to secondary parents by name

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1619449 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 256 insertions, 8 deletions

diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Model.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Model.java
index 9f671b47a8..6f5bd2b405 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Model.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Model.java
@@ -98,7 +98,7 @@ public abstract class Model
         return null;
     }
 
-    private Class<? extends ConfiguredObject> getAncestorClassWithGivenDescendant(
+    public Class<? extends ConfiguredObject> getAncestorClassWithGivenDescendant(
             final Class<? extends ConfiguredObject> category,
             final Class<? extends ConfiguredObject> descendantClass)
     {
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/ConfiguredObjectRecordConverter.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/ConfiguredObjectRecordConverter.java
index cc284a33f4..5f1c0b4b7f 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/ConfiguredObjectRecordConverter.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/ConfiguredObjectRecordConverter.java
@@ -22,8 +22,11 @@ package org.apache.qpid.server.store;
 
 import java.io.IOException;
 import java.io.Reader;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 
@@ -37,6 +40,11 @@ public class ConfiguredObjectRecordConverter
 {
     private final Model _model;
 
+    private static interface NameToIdResolver
+    {
+        public boolean resolve(Map<UUID, ConfiguredObjectRecord> objectsById);
+    }
+
     public ConfiguredObjectRecordConverter(final Model model)
     {
         _model = model;
@@ -52,16 +60,32 @@ public class ConfiguredObjectRecordConverter
         Map data = objectMapper.readValue(reader, Map.class);
         if(!data.isEmpty())
         {
-            loadChild(rootClass, data, parent.getCategoryClass(), parent.getId(), objectsById);
+            Collection<NameToIdResolver> unresolved =
+                    loadChild(rootClass, data, parent.getCategoryClass(), parent.getId(), objectsById);
+
+            Iterator<NameToIdResolver> iterator = unresolved.iterator();
+            while(iterator.hasNext())
+            {
+                if(iterator.next().resolve(objectsById))
+                {
+                    iterator.remove();
+                }
+            }
+
+            if(!unresolved.isEmpty())
+            {
+                throw new IllegalArgumentException("Initial configuration has unresolved references");
+            }
         }
         return objectsById.values();
     }
 
 
-    private void loadChild(final Class<? extends ConfiguredObject> clazz,
-                           final Map<String, Object> data,
-                           final Class<? extends ConfiguredObject> parentClass,
-                           final UUID parentId, final Map<UUID, ConfiguredObjectRecord> records)
+    private Collection<NameToIdResolver> loadChild(final Class<? extends ConfiguredObject> clazz,
+                                                   final Map<String, Object> data,
+                                                   final Class<? extends ConfiguredObject> parentClass,
+                                                   final UUID parentId,
+                                                   final Map<UUID, ConfiguredObjectRecord> records)
     {
         String idStr = (String) data.remove("id");
 
@@ -70,6 +94,7 @@ public class ConfiguredObjectRecordConverter
         Map<String,UUID> parentMap = new HashMap<>();
 
         Collection<Class<? extends ConfiguredObject>> childClasses = _model.getChildTypes(clazz);
+        List<NameToIdResolver> requiringResolution = new ArrayList<>();
         for(Class<? extends ConfiguredObject> childClass : childClasses)
         {
             final String childType = childClass.getSimpleName();
@@ -83,13 +108,14 @@ public class ConfiguredObjectRecordConverter
                     {
                         if(child instanceof Map)
                         {
-                            loadChild(childClass, (Map)child, clazz, id, records);
+                            requiringResolution.addAll(loadChild(childClass, (Map) child, clazz, id, records));
                         }
                     }
                 }
             }
 
         }
+
         if(parentId != null)
         {
             parentMap.put(parentClass.getSimpleName(),parentId);
@@ -107,7 +133,15 @@ public class ConfiguredObjectRecordConverter
                         }
                         catch(IllegalArgumentException e)
                         {
-                            // TODO
+                            final String ancestorClassName =
+                                    _model.getAncestorClassWithGivenDescendant(clazz, otherParent).getSimpleName();
+                            final String parentName = (String) otherParentId;
+                            final String parentType = otherParent.getSimpleName();
+
+                            requiringResolution.add(new AncestorFindingResolver(id,
+                                                                                parentType,
+                                                                                parentName,
+                                                                                ancestorClassName));
                         }
                     }
                 }
@@ -117,7 +151,79 @@ public class ConfiguredObjectRecordConverter
 
         records.put(id, new ConfiguredObjectRecordImpl(id, type, data, parentMap));
 
+        return requiringResolution;
     }
 
 
+    private static class AncestorFindingResolver implements NameToIdResolver
+    {
+        private final String _parentType;
+        private final String _parentName;
+        private final String _commonAncestorType;
+        private final UUID _id;
+
+        public AncestorFindingResolver(final UUID id,
+                                       final String parentType,
+                                       final String parentName,
+                                       final String commonAncestorType)
+        {
+            _id = id;
+            _parentType = parentType;
+            _parentName = parentName;
+            _commonAncestorType = commonAncestorType;
+        }
+
+        @Override
+        public boolean resolve(final Map<UUID, ConfiguredObjectRecord> objectsById)
+        {
+
+            ConfiguredObjectRecord record = objectsById.get(_id);
+            Collection<ConfiguredObjectRecord> recordsWithMatchingName = new ArrayList<>();
+            for(ConfiguredObjectRecord possibleParentRecord : objectsById.values())
+            {
+                if(possibleParentRecord.getType().equals(_parentType)
+                   && _parentName.equals(possibleParentRecord.getAttributes().get(ConfiguredObject.NAME)))
+                {
+                    recordsWithMatchingName.add(possibleParentRecord);
+                }
+            }
+            for(ConfiguredObjectRecord candidate : recordsWithMatchingName)
+            {
+                UUID candidateAncestor = findAncestor(candidate, _commonAncestorType, objectsById);
+                UUID recordAncestor = findAncestor(record, _commonAncestorType, objectsById);
+                if(recordAncestor.equals(candidateAncestor))
+                {
+                    HashMap<String, UUID> parents = new HashMap<>(record.getParents());
+                    parents.put(_parentType, candidate.getId());
+                    objectsById.put(_id, new ConfiguredObjectRecordImpl(_id, record.getType(), record.getAttributes(), parents));
+
+                    return true;
+                }
+            }
+            return false;
+        }
+
+        private UUID findAncestor(final ConfiguredObjectRecord record,
+                                  final String commonAncestorType,
+                                  final Map<UUID, ConfiguredObjectRecord> objectsById)
+        {
+            UUID id = record.getParents().get(commonAncestorType);
+            if(id == null)
+            {
+                for(UUID parentId : record.getParents().values())
+                {
+                    ConfiguredObjectRecord parent = objectsById.get(parentId);
+                    if(parent != null)
+                    {
+                        id = findAncestor(parent, commonAncestorType, objectsById);
+                    }
+                    if(id != null)
+                    {
+                        break;
+                    }
+                }
+            }
+            return id;
+        }
+    }
 }
diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/ConfiguredObjectRecordConverterTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/ConfiguredObjectRecordConverterTest.java
new file mode 100644
index 0000000000..bef3cdcac9
--- /dev/null
+++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/ConfiguredObjectRecordConverterTest.java
@@ -0,0 +1,142 @@
+/*
+ *
+ * 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.server.store;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.StringReader;
+import java.util.Collection;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.qpid.server.model.Binding;
+import org.apache.qpid.server.model.BrokerModel;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.model.VirtualHostNode;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class ConfiguredObjectRecordConverterTest extends QpidTestCase
+{
+
+    public void testSecondParentReferencedByName() throws Exception
+    {
+
+        String jsonData = "{\n"
+                          + "  \"name\" : \"test\",\n"
+                          + "  \"exchanges\" : [ {\n"
+                          + "    \"name\" : \"amq.direct\",\n"
+                          + "    \"type\" : \"direct\"\n"
+                          + "  } ],\n"
+                          + "  \"queues\" : [ {\n"
+                          + "    \"name\" : \"foo\",\n"
+                          + "    \"bindings\" : [ {\n"
+                          + "      \"exchange\" : \"amq.direct\",\n"
+                          + "      \"name\" : \"foo\"\n"
+                          + "    } ]\n"
+                          + "  } ]\n"
+                          + "} ";
+
+        ConfiguredObjectRecordConverter converter = new ConfiguredObjectRecordConverter(BrokerModel.getInstance());
+        ConfiguredObject parent = mock(ConfiguredObject.class);
+        when(parent.getId()).thenReturn(UUID.randomUUID());
+        when(parent.getCategoryClass()).thenReturn(VirtualHostNode.class);
+        Collection<ConfiguredObjectRecord> records =
+                converter.readFromJson(VirtualHost.class, parent, new StringReader(jsonData));
+
+        UUID exchangeId = null;
+        for (ConfiguredObjectRecord record : records)
+        {
+            if (record.getType().equals(Exchange.class.getSimpleName()))
+            {
+                assertNull("Only one exchange record expected", exchangeId);
+                exchangeId = record.getId();
+            }
+        }
+        assertNotNull("No exchange record found", exchangeId);
+
+        UUID queueId = null;
+        for (ConfiguredObjectRecord record : records)
+        {
+            if (record.getType().equals(Queue.class.getSimpleName()))
+            {
+                assertNull("Only one queue record expected", queueId);
+                queueId = record.getId();
+            }
+        }
+        assertNotNull("No queueId record found", queueId);
+
+        boolean bindingFound = false;
+        for (ConfiguredObjectRecord record : records)
+        {
+            if (record.getType().equals(Binding.class.getSimpleName()))
+            {
+                assertFalse("Expecting only one binding", bindingFound);
+                bindingFound = true;
+                Map<String,UUID> parents = record.getParents();
+                assertEquals("Two parents expected", 2, parents.size());
+                assertEquals("Queue parent id not as expected", queueId, parents.get(Queue.class.getSimpleName()));
+                assertEquals("Exchange parent id not as expected", exchangeId, parents.get(Exchange.class.getSimpleName()));
+
+            }
+        }
+        assertTrue("No binding found", bindingFound);
+    }
+
+    public void testUnresolvedSecondParentFailsToCovert() throws Exception
+    {
+        {
+
+            String jsonData = "{\n"
+                              + "  \"name\" : \"test\",\n"
+                              + "  \"exchanges\" : [ {\n"
+                              + "    \"name\" : \"amq.direct\",\n"
+                              + "    \"type\" : \"direct\"\n"
+                              + "  } ],\n"
+                              + "  \"queues\" : [ {\n"
+                              + "    \"name\" : \"foo\",\n"
+                              + "    \"bindings\" : [ {\n"
+                              + "      \"exchange\" : \"amq.topic\",\n"
+                              + "      \"name\" : \"foo\"\n"
+                              + "    } ]\n"
+                              + "  } ]\n"
+                              + "} ";
+
+            ConfiguredObjectRecordConverter converter = new ConfiguredObjectRecordConverter(BrokerModel.getInstance());
+            ConfiguredObject parent = mock(ConfiguredObject.class);
+            when(parent.getId()).thenReturn(UUID.randomUUID());
+            when(parent.getCategoryClass()).thenReturn(VirtualHostNode.class);
+            try
+            {
+                converter.readFromJson(VirtualHost.class, parent, new StringReader(jsonData));
+                fail("The records should not be converted as there is an unresolved reference");
+            }
+            catch (IllegalArgumentException e)
+            {
+                // pass
+            }
+
+        }
+    }
+}