/*
 *
 * 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.management.amqp;

import java.nio.charset.Charset;
import java.security.AccessControlException;
import java.security.AccessController;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;

import javax.security.auth.Subject;

import org.apache.qpid.server.connection.SessionPrincipal;
import org.apache.qpid.server.consumer.ConsumerTarget;
import org.apache.qpid.server.filter.FilterManager;
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.AcquiringMessageInstanceConsumer;
import org.apache.qpid.server.message.BaseMessageInstance;
import org.apache.qpid.server.message.ConsumerOption;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.message.internal.InternalMessageHeader;
import org.apache.qpid.server.model.AbstractConfigurationChangeListener;
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.ManagedObject;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.plugin.MessageConverter;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.MessageConverterRegistry;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.store.MessageDurability;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.store.TransactionLogResource;
import org.apache.qpid.server.txn.AutoCommitTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.StateChangeListener;

class ManagementNode implements MessageSource<ManagementNodeConsumer>, MessageDestination
{

    public static final String NAME_ATTRIBUTE = "name";
    public static final String IDENTITY_ATTRIBUTE = "identity";
    public static final String TYPE_ATTRIBUTE = "type";
    public static final String OPERATION_HEADER = "operation";
    public static final String SELF_NODE_NAME = "self";
    public static final String MANAGEMENT_TYPE = "org.amqp.management";
    public static final String GET_TYPES = "GET-TYPES";
    public static final String GET_ATTRIBUTES = "GET-ATTRIBUTES";
    public static final String GET_OPERATIONS = "GET-OPERATIONS";
    public static final String QUERY = "QUERY";
    public static final String ENTITY_TYPE_HEADER = "entityType";
    public static final String STATUS_CODE_HEADER = "statusCode";
    public static final int STATUS_CODE_OK = 200;
    public static final String ATTRIBUTES_HEADER = "attributes";
    public static final String OFFSET_HEADER = "offset";
    public static final String COUNT_HEADER = "count";
    public static final String MANAGEMENT_NODE_NAME = "$management";
    public static final String CREATE_OPERATION = "CREATE";
    public static final String READ_OPERATION = "READ";
    public static final String UPDATE_OPERATION = "UPDATE";
    public static final String DELETE_OPERATION = "DELETE";
    public static final String STATUS_DESCRIPTION_HEADER = "statusDescription";
    public static final int NOT_FOUND_STATUS_CODE = 404;
    public static final int NOT_IMPLEMENTED_STATUS_CODE = 501;
    public static final int STATUS_CODE_NO_CONTENT = 204;
    public static final int STATUS_CODE_FORBIDDEN = 403;
    public static final int STATUS_CODE_BAD_REQUEST = 400;
    public static final int STATUS_CODE_INTERNAL_ERROR = 500;
    public static final String ATTRIBUTE_NAMES = "attributeNames";
    public static final String RESULTS = "results";


    private final NamedAddressSpace _addressSpace;

    private final UUID _id;

    private final Action<ManagementNode> _onDelete;
    private final ConfiguredObject<?> _managedObject;
    private List<ManagementNodeConsumer> _consumers = new CopyOnWriteArrayList<>();

    private Map<String,ManagedEntityType> _entityTypes = Collections.synchronizedMap(new LinkedHashMap<String, ManagedEntityType>());

    private Map<ManagedEntityType,Map<String,ConfiguredObject>> _entities = Collections.synchronizedMap(new LinkedHashMap<ManagedEntityType,Map<String,ConfiguredObject>>());


    public ManagementNode(final NamedAddressSpace addressSpace,
                          final ConfiguredObject<?> configuredObject,
                          final Action<ManagementNode> onDelete)
    {
        _addressSpace = addressSpace;
        _onDelete = onDelete;
        final String name = configuredObject.getId() + MANAGEMENT_NODE_NAME;
        _id = UUID.nameUUIDFromBytes(name.getBytes(Charset.defaultCharset()));


        _managedObject = configuredObject;

        configuredObject.addChangeListener(new ModelObjectListener());
    }

    private Class getManagementClass(Class objectClass)
    {

        if(objectClass.getAnnotation(ManagedObject.class)!=null)
        {
            return objectClass;
        }
        List<Class> allClasses = Collections.singletonList(objectClass);
        List<Class> testedClasses = new ArrayList<Class>();
        do
        {
            testedClasses.addAll( allClasses );
            allClasses = new ArrayList<Class>();
            for(Class c : testedClasses)
            {
                for(Class i : c.getInterfaces())
                {
                    if(!allClasses.contains(i))
                    {
                        allClasses.add(i);
                    }
                }
                if(c.getSuperclass() != null && !allClasses.contains(c.getSuperclass()))
                {
                    allClasses.add(c.getSuperclass());
                }
            }
            allClasses.removeAll(testedClasses);
            for(Class c : allClasses)
            {
                if(c.getAnnotation(ManagedObject.class) != null)
                {
                    return c;
                }
            }
        }
        while(!allClasses.isEmpty());
        return null;
    }

    private boolean populateTypeMetaData(final Class<? extends ConfiguredObject> objectClass, boolean allowCreate)
    {
        Class clazz = getManagementClass(objectClass);
        if( clazz != null)
        {
            ManagedObject annotation = (ManagedObject) clazz.getAnnotation(ManagedObject.class);
            populateTypeMetaData(clazz, annotation);
            return true;
        }
        else
        {
            return false;
        }
    }

    private ManagedEntityType populateTypeMetaData(Class clazz,
                                                   final ManagedObject entityType)
    {

        ManagedEntityType managedEntityType = _entityTypes.get(clazz.getName());

        if(managedEntityType == null)
        {
            List<String> opsList = new ArrayList<String>(Arrays.asList(entityType.operations()));
            if(entityType.creatable())
            {
                boolean isCreatableChild = false;
                Collection<Class<? extends ConfiguredObject>> parentTypes = _managedObject.getModel().getParentTypes(clazz);
                for(Class<? extends ConfiguredObject> parentConfig : parentTypes)
                {
                    isCreatableChild = parentConfig.isAssignableFrom(_managedObject.getClass());
                    if(isCreatableChild)
                    {
                        opsList.add(CREATE_OPERATION);
                        break;
                    }
                }
            }
            opsList.addAll(Arrays.asList(READ_OPERATION, UPDATE_OPERATION, DELETE_OPERATION));

            Set<ManagedEntityType> parentSet = new HashSet<ManagedEntityType>();

            List<Class> allClasses = new ArrayList<Class>(Arrays.asList(clazz.getInterfaces()));
            if(clazz.getSuperclass() != null)
            {
                allClasses.add(clazz.getSuperclass());
            }

            for(Class parentClazz : allClasses)
            {
                if(parentClazz.getAnnotation(ManagedObject.class) != null)
                {
                    ManagedEntityType parentType = populateTypeMetaData(parentClazz,
                                                                        (ManagedObject) parentClazz.getAnnotation(
                                                                                ManagedObject.class)
                                                                       );
                    parentSet.add(parentType);
                    parentSet.addAll(Arrays.asList(parentType.getParents()));

                }
            }
            managedEntityType = new ManagedEntityType(clazz.getName(), parentSet.toArray(new ManagedEntityType[parentSet.size()]),
                                                      (String[])(_managedObject.getModel().getTypeRegistry().getAttributeNames(
                                                              clazz).toArray(new String[0])),
                                                      opsList.toArray(new String[opsList.size()]));
            _entityTypes.put(clazz.getName(),managedEntityType);
            _entities.put(managedEntityType, Collections.synchronizedMap(new LinkedHashMap<String, ConfiguredObject>()));

            if(ConfiguredObject.class.isAssignableFrom(clazz))
            {
                Collection<Class<? extends ConfiguredObject>> childTypes = _managedObject.getModel().getChildTypes(clazz);
                for(Class<? extends ConfiguredObject> childClass : childTypes)
                {
                    populateTypeMetaData(childClass, true);
                }
            }

        }

        return managedEntityType;

    }

    @Override
    public  <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
                                                                                  final String routingAddress,
                                                                                  final InstanceProperties instanceProperties,
                                                                                  final ServerTransaction txn,
                                                                                  final Action<? super BaseMessageInstance> postEnqueueAction)
    {

        @SuppressWarnings("unchecked")
        MessageConverter converter =
                MessageConverterRegistry.getConverter(message.getClass(), InternalMessage.class);

        final InternalMessage msg = (InternalMessage) converter.convert(message, _addressSpace);

        if(validateMessage(msg))
        {
            txn.addPostTransactionAction(new ServerTransaction.Action()
            {
                @Override
                public void postCommit()
                {
                    enqueue(msg, instanceProperties, postEnqueueAction);
                }

                @Override
                public void onRollback()
                {

                }
            });

            return 1;
        }
        else
        {
            return 0;
        }
    }

    private boolean validateMessage(final ServerMessage message)
    {
        AMQMessageHeader header = message.getMessageHeader();
        return containsStringHeader(header, TYPE_ATTRIBUTE) && containsStringHeader(header, OPERATION_HEADER)
               && (containsStringHeader(header, NAME_ATTRIBUTE) || containsStringHeader(header, IDENTITY_ATTRIBUTE));
    }

    private boolean containsStringHeader(final AMQMessageHeader header, String name)
    {
        return header.containsHeader(name) && header.getHeader(name) instanceof String;
    }

    synchronized void enqueue(InternalMessage message, InstanceProperties properties, Action<? super MessageInstance> postEnqueueAction)
    {
        if(postEnqueueAction != null)
        {
            postEnqueueAction.performAction(new ConsumedMessageInstance(message, properties));
        }


        String name = (String) message.getMessageHeader().getHeader(NAME_ATTRIBUTE);
        String id = (String) message.getMessageHeader().getHeader(IDENTITY_ATTRIBUTE);
        String type = (String) message.getMessageHeader().getHeader(TYPE_ATTRIBUTE);
        String operation = (String) message.getMessageHeader().getHeader(OPERATION_HEADER);

        InternalMessage response;

        if(SELF_NODE_NAME.equals(name) && type.equals(MANAGEMENT_TYPE))
        {
            response = performManagementOperation(message);
        }
        else if(CREATE_OPERATION.equals(operation))
        {
            response = performCreateOperation(message, type);
        }
        else
        {

            ConfiguredObject entity = findSubject(name, id, type);

            if(entity != null)
            {
                response = performOperation(message, entity);
            }
            else
            {
                if(id != null)
                {
                    response = createFailureResponse(message,
                                                     NOT_FOUND_STATUS_CODE,
                                                     "No entity with id {0} of type {1} found", id, type);
                }
                else
                {
                    response = createFailureResponse(message,
                                                     NOT_FOUND_STATUS_CODE,
                                                     "No entity with name {0} of type {1} found", name, type);
                }
            }
        }


        sendResponse(message, response);

    }

    private void sendResponse(final InternalMessage message, final InternalMessage response)
    {
        String replyTo = message.getMessageHeader().getReplyTo();
        response.setInitialRoutingAddress(replyTo);


        getResponseDestination(replyTo).send(response,
                                             replyTo, InstanceProperties.EMPTY,
                                             new AutoCommitTransaction(_addressSpace.getMessageStore()),
                                             null);

    }

    private MessageDestination getResponseDestination(String replyTo)
    {
        ManagementNodeConsumer consumer = null;
        Subject currentSubject = Subject.getSubject(AccessController.getContext());
        Set<SessionPrincipal> sessionPrincipals = currentSubject.getPrincipals(SessionPrincipal.class);
        if (!sessionPrincipals.isEmpty())
        {
            AMQSessionModel publishingSession = sessionPrincipals.iterator().next().getSession();
            for (ManagementNodeConsumer candidate : _consumers)
            {
                if (candidate.getTarget().getTargetAddress().equals(replyTo) && candidate.getSessionModel() == publishingSession)
                {
                    consumer = candidate;
                    break;
                }
            }
        }

        return consumer == null ? _addressSpace.getDefaultDestination() : consumer;
    }

    private InternalMessage performCreateOperation(final InternalMessage message, final String type)
    {
        InternalMessage response;
        ManagedEntityType entityType = _entityTypes.get(type);
        if(type != null)
        {
            if(Arrays.asList(entityType.getOperations()).contains(CREATE_OPERATION))
            {
                Object messageBody = message.getMessageBody();
                if(messageBody instanceof Map)
                {
                    try
                    {

                        Class<? extends ConfiguredObject> clazz =
                                (Class<? extends ConfiguredObject>) Class.forName(type);
                        try
                        {
                            ConfiguredObject child = _managedObject.createChild(clazz, (Map) messageBody);
                            if(child == null)
                            {
                                child = _entities.get(entityType).get(message.getMessageHeader().getHeader(NAME_ATTRIBUTE));
                            }
                            response = performReadOperation(message, child);
                        }
                        catch(AccessControlException e)
                        {
                            response = createFailureResponse(message, STATUS_CODE_FORBIDDEN, e.getMessage());
                        }
                    }
                    catch (ClassNotFoundException e)
                    {
                        response = createFailureResponse(message,
                                                         STATUS_CODE_INTERNAL_ERROR, "Unable to instantiate an instance of {0} ", type);
                    }
                }
                else
                {
                    response = createFailureResponse(message,
                                                     STATUS_CODE_BAD_REQUEST,
                                                     "The message body in the request was not of the correct type");
                }
            }
            else
            {
                response = createFailureResponse(message,
                                                 STATUS_CODE_FORBIDDEN,
                                                 "Cannot CREATE entities of type {0}", type);
            }
        }
        else
        {
            response = createFailureResponse(message,
                                             NOT_FOUND_STATUS_CODE,
                                             "Unknown type {0}",type);
        }
        return response;
    }

    private InternalMessage performOperation(final InternalMessage requestMessage, final ConfiguredObject entity)
    {
        String operation = (String) requestMessage.getMessageHeader().getHeader(OPERATION_HEADER);

        if(READ_OPERATION.equals(operation))
        {
            return performReadOperation(requestMessage, entity);
        }
        else if(DELETE_OPERATION.equals(operation))
        {
            return performDeleteOperation(requestMessage, entity);
        }
        else if(UPDATE_OPERATION.equals(operation))
        {
            return performUpdateOperation(requestMessage, entity);
        }
        else
        {
            return createFailureResponse(requestMessage, NOT_IMPLEMENTED_STATUS_CODE, "Unable to perform the {0} operation",operation);
        }
    }

    private InternalMessage performReadOperation(final InternalMessage requestMessage, final ConfiguredObject entity)
    {
        final InternalMessageHeader requestHeader = requestMessage.getMessageHeader();
        final MutableMessageHeader responseHeader = new MutableMessageHeader();
        responseHeader.setCorrelationId(requestHeader.getCorrelationId() == null
                                                ? requestHeader.getMessageId()
                                                : requestHeader.getCorrelationId());
        responseHeader.setMessageId(UUID.randomUUID().toString());
        responseHeader.setHeader(NAME_ATTRIBUTE, entity.getName());
        responseHeader.setHeader(IDENTITY_ATTRIBUTE, entity.getId().toString());
        responseHeader.setHeader(STATUS_CODE_HEADER,STATUS_CODE_OK);
        final String type = getManagementClass(entity.getClass()).getName();
        responseHeader.setHeader(TYPE_ATTRIBUTE, type);

        Map<String,Object> responseBody = new LinkedHashMap<String, Object>();
        final ManagedEntityType entityType = _entityTypes.get(type);
        for(String attribute : entityType.getAttributes())
        {
            responseBody.put(attribute, fixValue(entity.getAttribute(attribute)));
        }

        return InternalMessage.createMapMessage(_addressSpace.getMessageStore(), responseHeader, responseBody);
    }


    private InternalMessage performDeleteOperation(final InternalMessage requestMessage, final ConfiguredObject entity)
    {
        final InternalMessageHeader requestHeader = requestMessage.getMessageHeader();
        final MutableMessageHeader responseHeader = new MutableMessageHeader();
        responseHeader.setCorrelationId(requestHeader.getCorrelationId() == null
                                                ? requestHeader.getMessageId()
                                                : requestHeader.getCorrelationId());
        responseHeader.setMessageId(UUID.randomUUID().toString());
        responseHeader.setHeader(NAME_ATTRIBUTE, entity.getName());
        responseHeader.setHeader(IDENTITY_ATTRIBUTE, entity.getId().toString());
        final String type = getManagementClass(entity.getClass()).getName();
        responseHeader.setHeader(TYPE_ATTRIBUTE, type);
        try
        {
            entity.delete();
            responseHeader.setHeader(STATUS_CODE_HEADER, STATUS_CODE_NO_CONTENT);
        }
        catch(AccessControlException e)
        {
            responseHeader.setHeader(STATUS_CODE_HEADER, STATUS_CODE_FORBIDDEN);
        }

        return InternalMessage.createMapMessage(_addressSpace.getMessageStore(), responseHeader, Collections.emptyMap());
    }


    private InternalMessage performUpdateOperation(final InternalMessage requestMessage, final ConfiguredObject entity)
    {
        final InternalMessageHeader requestHeader = requestMessage.getMessageHeader();
        final MutableMessageHeader responseHeader = new MutableMessageHeader();
        responseHeader.setCorrelationId(requestHeader.getCorrelationId() == null
                                                ? requestHeader.getMessageId()
                                                : requestHeader.getCorrelationId());
        responseHeader.setMessageId(UUID.randomUUID().toString());
        responseHeader.setHeader(NAME_ATTRIBUTE, entity.getName());
        responseHeader.setHeader(IDENTITY_ATTRIBUTE, entity.getId().toString());
        final String type = getManagementClass(entity.getClass()).getName();
        responseHeader.setHeader(TYPE_ATTRIBUTE, type);

        Object messageBody = requestMessage.getMessageBody();
        if(messageBody instanceof Map)
        {
            try
            {
                entity.setAttributes((Map)messageBody);
                return performReadOperation(requestMessage, entity);
            }
            catch(AccessControlException e)
            {
                return createFailureResponse(requestMessage, STATUS_CODE_FORBIDDEN, e.getMessage());
            }
        }
        else
        {
            return createFailureResponse(requestMessage,
                                             STATUS_CODE_BAD_REQUEST,
                                             "The message body in the request was not of the correct type");
        }


    }

    private ConfiguredObject findSubject(final String name, final String id, final String type)
    {
        ConfiguredObject subject;
        ManagedEntityType met = _entityTypes.get(type);
        if(met == null)
        {
            return null;
        }

        subject = findSubject(name, id, met);
        if(subject == null)
        {
            ArrayList<ManagedEntityType> allTypes = new ArrayList<ManagedEntityType>(_entityTypes.values());
            for(ManagedEntityType entityType : allTypes)
            {
                if(Arrays.asList(entityType.getParents()).contains(met))
                {
                    subject = findSubject(name, id, entityType);
                    if(subject != null)
                    {
                        return subject;
                    }
                }
            }
        }
        return subject;
    }

    private ConfiguredObject findSubject(final String name, final String id, final ManagedEntityType entityType)
    {

        Map<String, ConfiguredObject> objects = _entities.get(entityType);
        if(name != null)
        {
            ConfiguredObject subject = objects.get(name);
            if(subject != null)
            {
                return subject;
            }
        }
        else
        {
            final Collection<ConfiguredObject> values = new ArrayList<ConfiguredObject>(objects.values());
            for(ConfiguredObject o : values)
            {
                if(o.getId().toString().equals(id))
                {
                    return o;
                }
            }
        }
        return null;
    }

    private InternalMessage createFailureResponse(final InternalMessage requestMessage,
                                       final int statusCode,
                                       final String stateDescription,
                                       Object... params)
    {
        final InternalMessageHeader requestHeader = requestMessage.getMessageHeader();
        final MutableMessageHeader responseHeader = new MutableMessageHeader();
        responseHeader.setCorrelationId(requestHeader.getCorrelationId() == null
                                                ? requestHeader.getMessageId()
                                                : requestHeader.getCorrelationId());
        responseHeader.setMessageId(UUID.randomUUID().toString());
        for(String header : requestHeader.getHeaderNames())
        {
            responseHeader.setHeader(header, requestHeader.getHeader(header));
        }
        responseHeader.setHeader(STATUS_CODE_HEADER, statusCode);
        responseHeader.setHeader(STATUS_DESCRIPTION_HEADER, MessageFormat.format(stateDescription, params));
        return InternalMessage.createBytesMessage(_addressSpace.getMessageStore(), responseHeader, new byte[0]);

    }

    private InternalMessage performManagementOperation(final InternalMessage msg)
    {
        final InternalMessage responseMessage;
        final InternalMessageHeader requestHeader = msg.getMessageHeader();
        final MutableMessageHeader responseHeader = new MutableMessageHeader();
        responseHeader.setCorrelationId(requestHeader.getCorrelationId() == null
                                                ? requestHeader.getMessageId()
                                                : requestHeader.getCorrelationId());
        responseHeader.setMessageId(UUID.randomUUID().toString());


        String operation = (String) requestHeader.getHeader(OPERATION_HEADER);
        if(GET_TYPES.equals(operation))
        {
            responseMessage = performGetTypes(requestHeader, responseHeader);
        }
        else if(GET_ATTRIBUTES.equals(operation))
        {
            responseMessage = performGetAttributes(requestHeader, responseHeader);
        }
        else if(GET_OPERATIONS.equals(operation))
        {
            responseMessage = performGetOperations(requestHeader, responseHeader);
        }
        else if(QUERY.equals(operation))
        {
            responseMessage = performQuery(requestHeader, msg.getMessageBody(), responseHeader);
        }
        else
        {
            responseMessage = InternalMessage.createBytesMessage(_addressSpace.getMessageStore(), requestHeader, new byte[0]);
        }
        return responseMessage;
    }

    private InternalMessage performGetTypes(final InternalMessageHeader requestHeader,
                                            final MutableMessageHeader responseHeader)
    {
        final InternalMessage responseMessage;
        List<String> restriction;
        if(requestHeader.containsHeader(ENTITY_TYPE_HEADER))
        {
            restriction = new ArrayList<String>(Collections.singletonList( (String)requestHeader.getHeader(ENTITY_TYPE_HEADER)));
        }
        else
        {
            restriction = null;
        }

        responseHeader.setHeader(STATUS_CODE_HEADER, STATUS_CODE_OK);
        Map<String,Object> responseMap = new LinkedHashMap<String, Object>();
        Map<String,ManagedEntityType> entityMapCopy;
        synchronized (_entityTypes)
        {
            entityMapCopy = new LinkedHashMap<String, ManagedEntityType>(_entityTypes);
        }

        for(ManagedEntityType type : entityMapCopy.values())
        {
            if(restriction == null || meetsIndirectRestriction(type,restriction))
            {
                final ManagedEntityType[] parents = type.getParents();
                List<String> parentNames = new ArrayList<String>();
                if(parents != null)
                {
                    for(ManagedEntityType parent : parents)
                    {
                        parentNames.add(parent.getName());
                    }
                }
                responseMap.put(type.getName(), parentNames);
            }
        }
        responseMessage = InternalMessage.createMapMessage(_addressSpace.getMessageStore(), responseHeader, responseMap);
        return responseMessage;
    }

    private InternalMessage performGetAttributes(final InternalMessageHeader requestHeader,
                                            final MutableMessageHeader responseHeader)
    {
        final InternalMessage responseMessage;
        String restriction;
        if(requestHeader.containsHeader(ENTITY_TYPE_HEADER))
        {
            restriction = (String) requestHeader.getHeader(ENTITY_TYPE_HEADER);
        }
        else
        {
            restriction = null;
        }

        responseHeader.setHeader(STATUS_CODE_HEADER, STATUS_CODE_OK);
        Map<String,Object> responseMap = new LinkedHashMap<String, Object>();
        Map<String,ManagedEntityType> entityMapCopy;
        synchronized (_entityTypes)
        {
            entityMapCopy = new LinkedHashMap<String, ManagedEntityType>(_entityTypes);
        }

        if(restriction == null)
        {
            for(ManagedEntityType type : entityMapCopy.values())
            {
                responseMap.put(type.getName(), Arrays.asList(type.getAttributes()));
            }
        }
        else if(entityMapCopy.containsKey(restriction))
        {
            responseMap.put(restriction, Arrays.asList(entityMapCopy.get(restriction).getAttributes()));
        }

        responseMessage = InternalMessage.createMapMessage(_addressSpace.getMessageStore(), responseHeader, responseMap);
        return responseMessage;
    }


    private InternalMessage performGetOperations(final InternalMessageHeader requestHeader,
                                                 final MutableMessageHeader responseHeader)
    {
        final InternalMessage responseMessage;
        String restriction;
        if(requestHeader.containsHeader(ENTITY_TYPE_HEADER))
        {
            restriction = (String) requestHeader.getHeader(ENTITY_TYPE_HEADER);
        }
        else
        {
            restriction = null;
        }

        responseHeader.setHeader(STATUS_CODE_HEADER, STATUS_CODE_OK);
        Map<String,Object> responseMap = new LinkedHashMap<String, Object>();
        Map<String,ManagedEntityType> entityMapCopy;
        synchronized (_entityTypes)
        {
            entityMapCopy = new LinkedHashMap<String, ManagedEntityType>(_entityTypes);
        }

        if(restriction == null)
        {
            for(ManagedEntityType type : entityMapCopy.values())
            {
                responseMap.put(type.getName(), Arrays.asList(type.getOperations()));
            }
        }
        else if(entityMapCopy.containsKey(restriction))
        {
            ManagedEntityType type = entityMapCopy.get(restriction);
            responseMap.put(type.getName(), Arrays.asList(type.getOperations()));
        }
        responseMessage = InternalMessage.createMapMessage(_addressSpace.getMessageStore(), responseHeader, responseMap);
        return responseMessage;
    }

    private InternalMessage performQuery(final InternalMessageHeader requestHeader,
                                         final Object messageBody, final MutableMessageHeader responseHeader)
    {
        final InternalMessage responseMessage;
        List<String> restriction;
        List<String> attributes;
        int offset;
        int count;

        if(requestHeader.containsHeader(ENTITY_TYPE_HEADER))
        {
            restriction = new ArrayList<String>(Collections.singletonList((String) requestHeader.getHeader(
                    ENTITY_TYPE_HEADER)));
            responseHeader.setHeader(ENTITY_TYPE_HEADER, restriction);
        }
        else
        {
            restriction = new ArrayList<String>(_entityTypes.keySet());
        }


        if(messageBody instanceof Map && ((Map)messageBody).get(ATTRIBUTE_NAMES) instanceof List)
        {
            attributes = (List<String>) ((Map)messageBody).get(ATTRIBUTE_NAMES);
        }
        else
        {
            LinkedHashMap<String,Void> attributeSet = new LinkedHashMap<String, Void>();
            for(String entityType : restriction)
            {
                ManagedEntityType type = _entityTypes.get(entityType);
                if(type != null)
                {
                    for(String attributeName : type.getAttributes())
                    {
                        attributeSet.put(attributeName, null);
                    }
                }
            }
            attributes = new ArrayList<String>(attributeSet.keySet());

        }

        if(requestHeader.containsHeader(OFFSET_HEADER))
        {
            offset = ((Number) requestHeader.getHeader(OFFSET_HEADER)).intValue();
            responseHeader.setHeader(OFFSET_HEADER,offset);
        }
        else
        {
            offset = 0;
        }

        if(requestHeader.containsHeader(COUNT_HEADER))
        {
            count = ((Number) requestHeader.getHeader(COUNT_HEADER)).intValue();
        }
        else
        {
            count = Integer.MAX_VALUE;
        }


        responseHeader.setHeader(STATUS_CODE_HEADER, STATUS_CODE_OK);
        List<List<? extends Object>> responseList = new ArrayList<List<? extends Object>>();
        int rowNo = 0;
        for(String type : restriction)
        {
            ManagedEntityType entityType = _entityTypes.get(type);
            if(entityType != null)
            {
                Map<String, ConfiguredObject> entityMap = _entities.get(entityType);
                if(entityMap  != null)
                {
                    List<ConfiguredObject> entities;
                    synchronized(entityMap)
                    {
                        entities = new ArrayList<ConfiguredObject>(entityMap.values());
                    }
                    for(ConfiguredObject entity : entities)
                    {
                        if(rowNo++ >= offset)
                        {
                            Object[] attrValue = new Object[attributes.size()];
                            int col = 0;
                            for(String attr : attributes)
                            {
                                Object value;
                                if(TYPE_ATTRIBUTE.equals(attr))
                                {
                                    value = entityType.getName();
                                }
                                else
                                {
                                    value = fixValue(entity.getAttribute(attr));
                                }
                                attrValue[col++] = value;
                            }
                            responseList.add(Arrays.asList(attrValue));
                        }
                        if(responseList.size()==count+1)
                        {
                            break;
                        }
                    }
                }
            }

            if(responseList.size()==count)
            {
                break;
            }
        }
        responseHeader.setHeader(COUNT_HEADER, responseList.size());
        Map<String,List> responseMap = new HashMap<String, List>();
        responseMap.put(ATTRIBUTE_NAMES, attributes);
        responseMap.put(RESULTS, responseList);
        responseMessage = InternalMessage.createMapMessage(_addressSpace.getMessageStore(),
                                                           responseHeader,
                                                           responseMap);
        return responseMessage;
    }

    private Object fixValue(final Object value)
    {
        Object fixedValue;
        if(value instanceof Enum)
        {
            fixedValue = value.toString();
        }
        else if(value instanceof Map)
        {
            Map<Object, Object> oldValue = (Map<Object, Object>) value;
            Map<Object, Object> newValue = new LinkedHashMap<Object, Object>();
            for(Map.Entry<Object, Object> entry : oldValue.entrySet())
            {
                newValue.put(fixValue(entry.getKey()),fixValue(entry.getValue()));
            }
            fixedValue = newValue;
        }
        else if(value instanceof Collection)
        {
            Collection oldValue = (Collection) value;
            List newValue = new ArrayList(oldValue.size());
            for(Object o : oldValue)
            {
                newValue.add(fixValue(o));
            }
            fixedValue = newValue;
        }
        else if(value != null && value.getClass().isArray() && !(value instanceof byte[]))
        {
            fixedValue = fixValue(Arrays.asList((Object[])value));
        }
        else
        {
            fixedValue = value;
        }
        return fixedValue;

    }


    private boolean meetsIndirectRestriction(final ManagedEntityType type, final List<String> restriction)
    {
        if(restriction.contains(type.getName()))
        {
            return true;
        }
        if(type.getParents() != null)
        {
            for(ManagedEntityType parent : type.getParents())
            {
                if(meetsIndirectRestriction(parent, restriction))
                {
                    return true;
                }
            }
        }
        return false;
    }

    @Override
    public synchronized  ManagementNodeConsumer addConsumer(final ConsumerTarget target,
                                                            final FilterManager filters,
                                                            final Class<? extends ServerMessage> messageClass,
                                                            final String consumerName,
                                                            final EnumSet<ConsumerOption> options,
                                                            final Integer priority)
    {

        final ManagementNodeConsumer managementNodeConsumer = new ManagementNodeConsumer(consumerName,this, target);
        target.consumerAdded(managementNodeConsumer);
        _consumers.add(managementNodeConsumer);
        target.addStateListener(new StateChangeListener<ConsumerTarget, ConsumerTarget.State>()
        {
            @Override
            public void stateChanged(final ConsumerTarget object,
                                     final ConsumerTarget.State oldState,
                                     final ConsumerTarget.State newState)
            {
                if(newState == ConsumerTarget.State.CLOSED)
                {
                    _consumers.remove(managementNodeConsumer);
                }
            }
        });
        return managementNodeConsumer;
    }

    @Override
    public synchronized Collection<ManagementNodeConsumer> getConsumers()
    {
        return Collections.unmodifiableCollection(_consumers);
    }

    @Override
    public boolean verifySessionAccess(final AMQSessionModel<?> session)
    {
        return true;
    }

    @Override
    public NamedAddressSpace getAddressSpace()
    {
        return _addressSpace;
    }


    @Override
    public void authorisePublish(final SecurityToken token, final Map<String, Object> arguments)
            throws AccessControlException
    {
        // ? special permissions to publish to the management node
    }

    @Override
    public String getName()
    {
        return MANAGEMENT_NODE_NAME;
    }

    @Override
    public UUID getId()
    {
        return _id;
    }

    @Override
    public MessageDurability getMessageDurability()
    {
        return MessageDurability.NEVER;
    }

    private class ConsumedMessageInstance implements MessageInstance
    {
        private final ServerMessage _message;
        private final InstanceProperties _properties;

        public ConsumedMessageInstance(final ServerMessage message,
                                       final InstanceProperties properties)
        {
            _message = message;
            _properties = properties;
        }

        @Override
        public int getDeliveryCount()
        {
            return 0;
        }

        @Override
        public void incrementDeliveryCount()
        {

        }

        @Override
        public void decrementDeliveryCount()
        {

        }

        @Override
        public void addStateChangeListener(final StateChangeListener<? super MessageInstance, EntryState> listener)
        {

        }

        @Override
        public boolean removeStateChangeListener(final StateChangeListener<? super MessageInstance, EntryState> listener)
        {
            return false;
        }


        @Override
        public boolean acquiredByConsumer()
        {
            return false;
        }

        @Override
        public MessageInstanceConsumer getAcquiringConsumer()
        {
            return null;
        }

        @Override
        public MessageEnqueueRecord getEnqueueRecord()
        {
            return null;
        }

        @Override
        public boolean isAcquiredBy(final MessageInstanceConsumer consumer)
        {
            return false;
        }

        @Override
        public boolean removeAcquisitionFromConsumer(final MessageInstanceConsumer consumer)
        {
            return false;
        }

        @Override
        public void setRedelivered()
        {

        }

        @Override
        public boolean isRedelivered()
        {
            return false;
        }

        @Override
        public AcquiringMessageInstanceConsumer<?,?> getDeliveredConsumer()
        {
            return null;
        }

        @Override
        public void reject()
        {

        }

        @Override
        public boolean isRejectedBy(final MessageInstanceConsumer consumer)
        {
            return false;
        }

        @Override
        public boolean getDeliveredToConsumer()
        {
            return true;
        }

        @Override
        public boolean expired()
        {
            return false;
        }

        @Override
        public boolean acquire(final MessageInstanceConsumer sub)
        {
            return false;
        }

        @Override
        public boolean makeAcquisitionUnstealable(final MessageInstanceConsumer consumer)
        {
            return false;
        }

        @Override
        public boolean makeAcquisitionStealable()
        {
            return false;
        }

        @Override
        public int getMaximumDeliveryCount()
        {
            return 0;
        }

        @Override
        public int routeToAlternate(final Action<? super BaseMessageInstance> action,
                                    final ServerTransaction txn)
        {
            return 0;
        }


        @Override
        public Filterable asFilterable()
        {
            return null;
        }

        @Override
        public boolean isAvailable()
        {
            return false;
        }

        @Override
        public boolean acquire()
        {
            return false;
        }

        @Override
        public boolean isAcquired()
        {
            return false;
        }

        @Override
        public void release()
        {

        }

        @Override
        public void release(final MessageInstanceConsumer release)
        {

        }

        @Override
        public boolean resend()
        {
            return false;
        }

        @Override
        public void delete()
        {

        }

        @Override
        public boolean isDeleted()
        {
            return false;
        }

        @Override
        public boolean isHeld()
        {
            return false;
        }

        @Override
        public ServerMessage getMessage()
        {
            return _message;
        }

        @Override
        public InstanceProperties getInstanceProperties()
        {
            return _properties;
        }

        @Override
        public TransactionLogResource getOwningResource()
        {
            return ManagementNode.this;
        }
    }

    private class ModelObjectListener extends AbstractConfigurationChangeListener
    {
        @Override
        public void stateChanged(final ConfiguredObject object, final State oldState, final State newState)
        {
            if(newState == State.DELETED)
            {
                if(_onDelete != null)
                {
                    _onDelete.performAction(ManagementNode.this);
                }
            }
            else if(newState == State.ACTIVE && object instanceof org.apache.qpid.server.model.VirtualHost)
            {
                populateTypeMetaData(object.getClass(), false);
                final Class managementClass = getManagementClass(_managedObject.getClass());
                _entities.get(_entityTypes.get(managementClass.getName())).put(_managedObject.getName(), _managedObject);

                Collection<Class<? extends ConfiguredObject>> childClasses = object.getModel().getChildTypes(managementClass);
                for(Class<? extends ConfiguredObject> childClass : childClasses)
                {
                    if(getManagementClass(childClass) != null)
                    {
                        for(ConfiguredObject child : _managedObject.getChildren(childClass))
                        {
                            _entities.get(_entityTypes.get(getManagementClass(childClass).getName())).put(child.getName(), child);
                        }
                    }
                }

            }
        }

        @Override
        public void childAdded(final ConfiguredObject object, final ConfiguredObject child)
        {
            final Class managementClass = getManagementClass(child.getClass());
            final ManagedEntityType entityType = _entityTypes.get(managementClass.getName());
            if(entityType != null)
            {
                _entities.get(entityType).put(child.getName(), child);
            }
        }

        @Override
        public void childRemoved(final ConfiguredObject object, final ConfiguredObject child)
        {
            final ManagedEntityType entityType = _entityTypes.get(getManagementClass(child.getClass()).getName());
            if(entityType != null)
            {
                _entities.get(entityType).remove(child.getName());
            }
        }
    }

    private static class MutableMessageHeader implements AMQMessageHeader
    {
        private final LinkedHashMap<String, Object> _headers = new LinkedHashMap<String, Object>();
        private String _correlationId;
        private long _expiration;
        private String _userId;
        private String _appId;
        private String _messageId;
        private String _mimeType;
        private String _encoding;
        private byte _priority;
        private long _timestamp;
        private long _notValidBefore;
        private String _type;
        private String _replyTo;

        public void setCorrelationId(final String correlationId)
        {
            _correlationId = correlationId;
        }

        public void setExpiration(final long expiration)
        {
            _expiration = expiration;
        }

        public void setUserId(final String userId)
        {
            _userId = userId;
        }

        public void setAppId(final String appId)
        {
            _appId = appId;
        }

        public void setMessageId(final String messageId)
        {
            _messageId = messageId;
        }

        public void setMimeType(final String mimeType)
        {
            _mimeType = mimeType;
        }

        public void setEncoding(final String encoding)
        {
            _encoding = encoding;
        }

        public void setPriority(final byte priority)
        {
            _priority = priority;
        }

        public void setTimestamp(final long timestamp)
        {
            _timestamp = timestamp;
        }

        public void setNotValidBefore(final long notValidBefore)
        {
            _notValidBefore = notValidBefore;
        }

        public void setType(final String type)
        {
            _type = type;
        }

        public void setReplyTo(final String replyTo)
        {
            _replyTo = replyTo;
        }

        public String getCorrelationId()
        {
            return _correlationId;
        }

        public long getExpiration()
        {
            return _expiration;
        }

        public String getUserId()
        {
            return _userId;
        }

        public String getAppId()
        {
            return _appId;
        }

        public String getMessageId()
        {
            return _messageId;
        }

        public String getMimeType()
        {
            return _mimeType;
        }

        public String getEncoding()
        {
            return _encoding;
        }

        public byte getPriority()
        {
            return _priority;
        }

        public long getTimestamp()
        {
            return _timestamp;
        }

        @Override
        public long getNotValidBefore()
        {
            return _notValidBefore;
        }

        public String getType()
        {
            return _type;
        }

        public String getReplyTo()
        {
            return _replyTo;
        }

        @Override
        public Object getHeader(final String name)
        {
            return _headers.get(name);
        }

        @Override
        public boolean containsHeaders(final Set<String> names)
        {
            return _headers.keySet().containsAll(names);
        }

        @Override
        public boolean containsHeader(final String name)
        {
            return _headers.containsKey(name);
        }

        @Override
        public Collection<String> getHeaderNames()
        {
            return Collections.unmodifiableCollection(_headers.keySet());
        }

        public void setHeader(String header, Object value)
        {
            _headers.put(header,value);
        }

    }
}