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

import java.security.AccessControlException;
import java.security.AccessController;
import java.security.Principal;
import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

import javax.security.auth.Subject;
import javax.security.auth.login.AccountNotFoundException;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.qpid.server.BrokerPrincipal;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.configuration.CommonProperties;
import org.apache.qpid.server.configuration.IllegalConfigurationException;
import org.apache.qpid.server.configuration.updater.TaskExecutor;
import org.apache.qpid.server.configuration.updater.TaskExecutorImpl;
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.logging.messages.BrokerMessages;
import org.apache.qpid.server.logging.messages.VirtualHostMessages;
import org.apache.qpid.server.model.preferences.Preference;
import org.apache.qpid.server.model.preferences.UserPreferences;
import org.apache.qpid.server.model.preferences.UserPreferencesImpl;
import org.apache.qpid.server.plugin.QpidServiceLoader;
import org.apache.qpid.server.plugin.SystemAddressSpaceCreator;
import org.apache.qpid.server.plugin.SystemNodeCreator;
import org.apache.qpid.server.security.AccessControl;
import org.apache.qpid.server.security.CompoundAccessControl;
import org.apache.qpid.server.security.Result;
import org.apache.qpid.server.security.SubjectFixedResultAccessControl;
import org.apache.qpid.server.security.SubjectFixedResultAccessControl.ResultCalculator;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.security.auth.SocketConnectionMetaData;
import org.apache.qpid.server.security.auth.SocketConnectionPrincipal;
import org.apache.qpid.server.security.auth.UsernamePrincipal;
import org.apache.qpid.server.security.auth.manager.SimpleAuthenticationManager;
import org.apache.qpid.server.security.group.GroupPrincipal;
import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.stats.StatisticsGatherer;
import org.apache.qpid.server.store.FileBasedSettings;
import org.apache.qpid.server.store.preferences.PreferenceRecord;
import org.apache.qpid.server.store.preferences.PreferenceStore;
import org.apache.qpid.server.store.preferences.PreferenceStoreUpdaterImpl;
import org.apache.qpid.server.store.preferences.PreferencesRecoverer;
import org.apache.qpid.server.store.preferences.PreferencesRoot;
import org.apache.qpid.server.util.HousekeepingExecutor;
import org.apache.qpid.server.util.SystemUtils;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
import org.apache.qpid.server.virtualhost.VirtualHostPropertiesNodeCreator;

@ManagedObject( category = false, type = "Broker" )
public class BrokerImpl extends AbstractContainer<BrokerImpl> implements Broker<BrokerImpl>
{
    private static final Logger LOGGER = LoggerFactory.getLogger(BrokerImpl.class);

    private static final Pattern MODEL_VERSION_PATTERN = Pattern.compile("^\\d+\\.\\d+$");

    private static final int HOUSEKEEPING_SHUTDOWN_TIMEOUT = 5;

    public static final String MANAGEMENT_MODE_AUTHENTICATION = "MANAGEMENT_MODE_AUTHENTICATION";

    private final AccessControl _systemUserAllowed = new SubjectFixedResultAccessControl(new ResultCalculator()
    {
        @Override
        public Result getResult(final Subject subject)
        {
            return isSystemSubject(subject) ? Result.ALLOWED : Result.DEFER;
        }
    }, Result.DEFER);

    private final BrokerPrincipal _principal;

    private String[] POSITIVE_NUMERIC_ATTRIBUTES = { CONNECTION_SESSION_COUNT_LIMIT,
            CONNECTION_HEART_BEAT_DELAY, STATISTICS_REPORTING_PERIOD };


    private AuthenticationProvider<?> _managementModeAuthenticationProvider;

    private Timer _reportingTimer;
    private final StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;

    /** Flags used to control the reporting of flow to disk. Protected by this */
    private boolean _totalMessageSizeExceedThresholdReported = false,  _totalMessageSizeWithinThresholdReported = true;

    @ManagedAttributeField
    private int _connection_sessionCountLimit;
    @ManagedAttributeField
    private int _connection_heartBeatDelay;
    @ManagedAttributeField
    private boolean _connection_closeWhenNoRoute;
    @ManagedAttributeField
    private int _statisticsReportingPeriod;
    @ManagedAttributeField
    private boolean _statisticsReportingResetEnabled;
    @ManagedAttributeField
    private boolean _messageCompressionEnabled;

    private PreferenceStore _preferenceStore;

    private final boolean _virtualHostPropertiesNodeEnabled;
    private Collection<BrokerLogger> _brokerLoggersToClose;
    private int _networkBufferSize = DEFAULT_NETWORK_BUFFER_SIZE;
    private final AddressSpaceRegistry _addressSpaceRegistry = new AddressSpaceRegistry();
    private ConfigurationChangeListener _accessControlProviderListener = new AccessControlProviderListener();
    private final AccessControl _accessControl;
    private TaskExecutor _preferenceTaskExecutor;
    private String _documentationUrl;
    private volatile long _smallestAllowedBufferId = QpidByteBuffer.getLargestPooledBufferId();
    private long _compactMemoryThreshold;
    private long _compactMemoryInterval;
    private double _memoryOccupancyThreshold;
    private long _memoryCompactionIncrement;
    private long _flowToDiskThreshold;

    @ManagedObjectFactoryConstructor
    public BrokerImpl(Map<String, Object> attributes,
                      SystemConfig parent)
    {
        super(attributes, parent);
        _principal = new BrokerPrincipal(this);

        if (parent.isManagementMode())
        {
            Map<String,Object> authManagerAttrs = new HashMap<String, Object>();
            authManagerAttrs.put(NAME,"MANAGEMENT_MODE_AUTHENTICATION");
            authManagerAttrs.put(ID, UUID.randomUUID());
            SimpleAuthenticationManager authManager = new SimpleAuthenticationManager(authManagerAttrs, this);
            authManager.addUser(SystemConfig.MANAGEMENT_MODE_USER_NAME, _parent.getManagementModePassword());
            _managementModeAuthenticationProvider = authManager;
            _accessControl = AccessControl.ALWAYS_ALLOWED;
        }
        else
        {
            _accessControl =  new CompoundAccessControl(Collections.<AccessControl<?>>emptyList(), Result.ALLOWED);
        }

        QpidServiceLoader qpidServiceLoader = new QpidServiceLoader();
        final Set<String> systemNodeCreatorTypes = qpidServiceLoader.getInstancesByType(SystemNodeCreator.class).keySet();
        _virtualHostPropertiesNodeEnabled = systemNodeCreatorTypes.contains(VirtualHostPropertiesNodeCreator.TYPE);
        _messagesDelivered = new StatisticsCounter("messages-delivered");
        _dataDelivered = new StatisticsCounter("bytes-delivered");
        _messagesReceived = new StatisticsCounter("messages-received");
        _dataReceived = new StatisticsCounter("bytes-received");


    }

    private void registerSystemAddressSpaces()
    {
        QpidServiceLoader qpidServiceLoader = new QpidServiceLoader();
        Iterable<SystemAddressSpaceCreator> factories = qpidServiceLoader.instancesOf(SystemAddressSpaceCreator.class);
        for(SystemAddressSpaceCreator creator : factories)
        {
            creator.register(_addressSpaceRegistry);
        }
    }


    @Override
    protected void postResolve()
    {
        super.postResolve();
        Integer networkBufferSize = getContextValue(Integer.class, NETWORK_BUFFER_SIZE);
        if (networkBufferSize == null || networkBufferSize < MINIMUM_NETWORK_BUFFER_SIZE)
        {
            throw new IllegalConfigurationException(NETWORK_BUFFER_SIZE + " is set to unacceptable value '" +
                    networkBufferSize + "'. Must be larger than " + MINIMUM_NETWORK_BUFFER_SIZE + ".");
        }
        _networkBufferSize = networkBufferSize;

        int poolSize = getContextValue(Integer.class, BROKER_DIRECT_BYTE_BUFFER_POOL_SIZE);

        QpidByteBuffer.initialisePool(_networkBufferSize, poolSize);
    }

    @Override
    protected void postResolveChildren()
    {
        super.postResolveChildren();

        final SystemConfig parent = (SystemConfig) getParent();
        Runnable task =  parent.getOnContainerResolveTask();
        if(task != null)
        {
            task.run();
        }
        addChangeListener(_accessControlProviderListener);
        for(AccessControlProvider aclProvider : getChildren(AccessControlProvider.class))
        {
            aclProvider.addChangeListener(_accessControlProviderListener);
        }
        _eventLogger.message(BrokerMessages.CONFIG(parent instanceof FileBasedSettings
                                                           ? ((FileBasedSettings) parent).getStorePath()
                                                           : "N/A"));

    }

    @Override
    public void onValidate()
    {
        super.onValidate();
        String modelVersion = (String) getActualAttributes().get(Broker.MODEL_VERSION);
        if (modelVersion == null)
        {
            deleted();
            throw new IllegalConfigurationException("Broker " + Broker.MODEL_VERSION + " must be specified");
        }

        if (!MODEL_VERSION_PATTERN.matcher(modelVersion).matches())
        {
            deleted();
            throw new IllegalConfigurationException("Broker " + Broker.MODEL_VERSION + " is specified in incorrect format: "
                                                    + modelVersion);
        }

        int versionSeparatorPosition = modelVersion.indexOf(".");
        String majorVersionPart = modelVersion.substring(0, versionSeparatorPosition);
        int majorModelVersion = Integer.parseInt(majorVersionPart);
        int minorModelVersion = Integer.parseInt(modelVersion.substring(versionSeparatorPosition + 1));

        if (majorModelVersion != BrokerModel.MODEL_MAJOR_VERSION || minorModelVersion > BrokerModel.MODEL_MINOR_VERSION)
        {
            deleted();
            throw new IllegalConfigurationException("The model version '" + modelVersion
                                                    + "' in configuration is incompatible with the broker model version '" + BrokerModel.MODEL_VERSION + "'");
        }

        if(!isDurable())
        {
            deleted();
            throw new IllegalArgumentException(getClass().getSimpleName() + " must be durable");
        }

    }

    @Override
    protected void validateChange(final ConfiguredObject<?> proxyForValidation, final Set<String> changedAttributes)
    {
        super.validateChange(proxyForValidation, changedAttributes);
        Broker updated = (Broker) proxyForValidation;
        if (changedAttributes.contains(MODEL_VERSION) && !BrokerModel.MODEL_VERSION.equals(updated.getModelVersion()))
        {
            throw new IllegalConfigurationException("Cannot change the model version");
        }

        for (String attributeName : POSITIVE_NUMERIC_ATTRIBUTES)
        {
            if(changedAttributes.contains(attributeName))
            {
                Number value = (Number) updated.getAttribute(attributeName);

                if (value != null && value.longValue() < 0)
                {
                    throw new IllegalConfigurationException(
                            "Only positive integer value can be specified for the attribute "
                            + attributeName);
                }
            }
        }
    }

    @Override
    protected void validateChildDelete(final ConfiguredObject<?> child)
    {
        super.validateChildDelete(child);
        if(child instanceof AccessControlProvider && getChildren(AccessControlProvider.class).size() == 1)
        {
            String categoryName = child.getCategoryClass().getSimpleName();
            throw new IllegalConfigurationException("The " + categoryName + " named '" + child.getName()
                                                    + "' cannot be deleted as at least one " + categoryName
                                                    + " must be present");
        }
    }

    @StateTransition( currentState = State.UNINITIALIZED, desiredState = State.ACTIVE )
    private ListenableFuture<Void> activate()
    {
        if(_parent.isManagementMode())
        {
            return doAfter(_managementModeAuthenticationProvider.openAsync(),
                    new Runnable()
                    {
                        @Override
                        public void run()
                        {
                            performActivation();
                        }
                    });
        }
        else
        {
            performActivation();
            return Futures.immediateFuture(null);
        }
    }

    @SuppressWarnings("unused")
    @StateTransition( currentState = {State.ACTIVE, State.ERRORED}, desiredState = State.STOPPED )
    private ListenableFuture<Void> doStop()
    {
        stopPreferenceTaskExecutor();
        closePreferenceStore();
        return Futures.immediateFuture(null);
    }

    private void closePreferenceStore()
    {
        PreferenceStore ps = _preferenceStore;
        if (ps != null)
        {
            ps.close();
        }
    }

    private void stopPreferenceTaskExecutor()
    {
        TaskExecutor preferenceTaskExecutor = _preferenceTaskExecutor;
        if (preferenceTaskExecutor != null)
        {
            preferenceTaskExecutor.stop();
        }
    }

    @Override
    public void initiateShutdown()
    {
        getEventLogger().message(BrokerMessages.OPERATION("initiateShutdown"));
        _parent.closeAsync();
    }

    @Override
    public Map<String, Object> extractConfig(final boolean includeSecureAttributes)
    {
        return (new ConfigurationExtractor()).extractConfig(this, includeSecureAttributes);
    }

    private void performActivation()
    {
        boolean hasBrokerAnyErroredChildren = false;

        List<ConfiguredObject<?>> failedChildren = new ArrayList<>();
        for (final Class<? extends ConfiguredObject> childClass : getModel().getChildTypes(getCategoryClass()))
        {
            final Collection<? extends ConfiguredObject> children = getChildren(childClass);
            if (children != null) {
                for (final ConfiguredObject<?> child : children)
                {
                    if (child.getState() == State.ERRORED )
                    {
                        hasBrokerAnyErroredChildren = true;
                        LOGGER.warn("Broker child object '{}' of type '{}' is {}",
                                child.getName(), childClass.getSimpleName(), State.ERRORED);
                        failedChildren.add(child);
                    }
                }
            }
        }

        if(!failedChildren.isEmpty())
        {
            getEventLogger().message(BrokerMessages.FAILED_CHILDREN(failedChildren.toString()));
        }

        _documentationUrl = getContextValue(String.class, QPID_DOCUMENTATION_URL);
        final boolean brokerShutdownOnErroredChild = getContextValue(Boolean.class,
                                                                     BROKER_FAIL_STARTUP_WITH_ERRORED_CHILD);
        if (!_parent.isManagementMode() && brokerShutdownOnErroredChild && hasBrokerAnyErroredChildren)
        {
            throw new IllegalStateException(String.format("Broker context variable %s is set and the broker has %s children",
                    BROKER_FAIL_STARTUP_WITH_ERRORED_CHILD, State.ERRORED));
        }
        updateAccessControl();

        initialiseStatisticsReporting();

        _houseKeepingTaskExecutor = new HousekeepingExecutor("broker-" + getName() + "-pool",
                                                             getHousekeepingThreadCount(),
                                                             getSystemTaskSubject("Housekeeping", _principal));

        _houseKeepingTaskExecutor.scheduleWithFixedDelay(this::checkDirectMemoryUsage,
                                                         _compactMemoryInterval,
                                                         _compactMemoryInterval,
                                                         TimeUnit.MILLISECONDS);

        final PreferenceStoreUpdaterImpl updater = new PreferenceStoreUpdaterImpl();
        final Collection<PreferenceRecord> preferenceRecords = _preferenceStore.openAndLoad(updater);
        _preferenceTaskExecutor = new TaskExecutorImpl("broker-" + getName() + "-preferences", null);
        _preferenceTaskExecutor.start();
        PreferencesRecoverer preferencesRecoverer = new PreferencesRecoverer(_preferenceTaskExecutor);
        preferencesRecoverer.recoverPreferences(this, preferenceRecords, _preferenceStore);

        if (isManagementMode())
        {
            _eventLogger.message(BrokerMessages.MANAGEMENT_MODE(SystemConfig.MANAGEMENT_MODE_USER_NAME,
                                                                _parent.getManagementModePassword()));
        }
        setState(State.ACTIVE);
    }

    private void checkDirectMemoryUsage()
    {
        if (_compactMemoryThreshold >= 0 && getUsedDirectMemorySize() > _compactMemoryThreshold)
        {
            compactMemory();
        }
    }

    private void initialiseStatisticsReporting()
    {
        long report = getStatisticsReportingPeriod() * 1000L; // convert to ms
        final boolean reset = getStatisticsReportingResetEnabled();

        /* add a timer task to report statistics if generation is enabled for broker or virtualhosts */
        if (report > 0L)
        {
            _reportingTimer = new Timer("Statistics-Reporting", true);
            StatisticsReportingTask task = new StatisticsReportingTask(reset, _eventLogger);
            _reportingTimer.scheduleAtFixedRate(task, report / 2, report);
        }
    }

    @Override
    public int getConnection_sessionCountLimit()
    {
        return _connection_sessionCountLimit;
    }

    @Override
    public int getConnection_heartBeatDelay()
    {
        return _connection_heartBeatDelay;
    }

    @Override
    public boolean getConnection_closeWhenNoRoute()
    {
        return _connection_closeWhenNoRoute;
    }

    @Override
    public int getStatisticsReportingPeriod()
    {
        return _statisticsReportingPeriod;
    }

    @Override
    public boolean getStatisticsReportingResetEnabled()
    {
        return _statisticsReportingResetEnabled;
    }

    @Override
    public boolean isMessageCompressionEnabled()
    {
        return _messageCompressionEnabled;
    }

    @Override
    public Collection<VirtualHostNode<?>> getVirtualHostNodes()
    {
        Collection children = getChildren(VirtualHostNode.class);
        return children;
    }

    public Collection<Port<?>> getPorts()
    {
        Collection children = getChildren(Port.class);
        return children;
    }

    public Collection<AuthenticationProvider<?>> getAuthenticationProviders()
    {
        Collection children = getChildren(AuthenticationProvider.class);
        return children;
    }

    @Override
    public synchronized void assignTargetSizes()
    {
        LOGGER.debug("Assigning target sizes based on total target {}", _flowToDiskThreshold);
        long totalSize = 0l;
        Collection<VirtualHostNode<?>> vhns = getVirtualHostNodes();
        Map<QueueManagingVirtualHost<?>, Long> vhs = new HashMap<>();
        for (VirtualHostNode<?> vhn : vhns)
        {
            VirtualHost<?> vh = vhn.getVirtualHost();
            if (vh instanceof QueueManagingVirtualHost)
            {
                QueueManagingVirtualHost<?> host = (QueueManagingVirtualHost<?>)vh;
                long totalQueueDepthBytes = host.getTotalQueueDepthBytes();
                vhs.put(host, totalQueueDepthBytes);
                totalSize += totalQueueDepthBytes;
            }
        }

        if (totalSize > _flowToDiskThreshold && !_totalMessageSizeExceedThresholdReported)
        {
            _eventLogger.message(BrokerMessages.FLOW_TO_DISK_ACTIVE(totalSize / 1024, _flowToDiskThreshold / 1024));
            _totalMessageSizeExceedThresholdReported = true;
            _totalMessageSizeWithinThresholdReported = false;
        }
        else if (totalSize <= _flowToDiskThreshold && !_totalMessageSizeWithinThresholdReported)
        {
            _eventLogger.message(BrokerMessages.FLOW_TO_DISK_INACTIVE(totalSize / 1024, _flowToDiskThreshold / 1024));
            _totalMessageSizeWithinThresholdReported = true;
            _totalMessageSizeExceedThresholdReported = false;
        }

        final long proportionalShare = (long) ((double) _flowToDiskThreshold / (double) vhs.size());
        for (Map.Entry<QueueManagingVirtualHost<?>, Long> entry : vhs.entrySet())
        {
            long virtualHostTotalQueueSize = entry.getValue();
            final long size;
            if (totalSize == 0)
            {
                size = proportionalShare;
            }
            else
            {
                long queueSizeBasedShare = (_flowToDiskThreshold * virtualHostTotalQueueSize) / (2 * totalSize);
                size = queueSizeBasedShare + (proportionalShare / 2);
            }

            if (LOGGER.isDebugEnabled())
            {
                LOGGER.debug("Assigning target size {} to vhost {}", size, entry.getKey());
            }
            entry.getKey().setTargetSize(size);
        }
    }

    @Override
    protected void onOpen()
    {
        super.onOpen();

        PreferencesRoot preferencesRoot = (SystemConfig) getParent();
        _preferenceStore = preferencesRoot.createPreferenceStore();

        getEventLogger().message(BrokerMessages.STARTUP(CommonProperties.getReleaseVersion(),
                                                        CommonProperties.getBuildVersion()));

        getEventLogger().message(BrokerMessages.PLATFORM(System.getProperty("java.vendor"),
                                                         System.getProperty("java.runtime.version",
                                                                            System.getProperty("java.version")),
                                                         SystemUtils.getOSName(),
                                                         SystemUtils.getOSVersion(),
                                                         SystemUtils.getOSArch(),
                                                         String.valueOf(getNumberOfCores())));

        long directMemory = getMaxDirectMemorySize();
        long heapMemory = Runtime.getRuntime().maxMemory();
        getEventLogger().message(BrokerMessages.MAX_MEMORY(heapMemory, directMemory));

        _flowToDiskThreshold = getContextValue(Long.class, BROKER_FLOW_TO_DISK_THRESHOLD);
        _compactMemoryThreshold = getContextValue(Long.class, Broker.COMPACT_MEMORY_THRESHOLD);
        _compactMemoryInterval = getContextValue(Long.class, Broker.COMPACT_MEMORY_INTERVAL);
        _memoryOccupancyThreshold = getContextValue(Double.class, Broker.MEMORY_OCCUPANCY_THRESHOLD);
        _memoryCompactionIncrement = getContextValue(Long.class, Broker.MEMORY_COMPACTION_INCREMENT);

        if (SystemUtils.getProcessPid() != null)
        {
            getEventLogger().message(BrokerMessages.PROCESS(SystemUtils.getProcessPid()));
        }

        registerSystemAddressSpaces();

        assignTargetSizes();
    }

    @Override
    public NamedAddressSpace getSystemAddressSpace(String name)
    {
        return _addressSpaceRegistry.getAddressSpace(name);
    }

    @Override
    public Collection<GroupProvider<?>> getGroupProviders()
    {
        Collection children = getChildren(GroupProvider.class);
        return children;
    }

    private ListenableFuture<VirtualHostNode> createVirtualHostNodeAsync(Map<String, Object> attributes)
            throws AccessControlException, IllegalArgumentException
    {

        return doAfter(getObjectFactory().createAsync(VirtualHostNode.class, attributes, this),
                       new CallableWithArgument<ListenableFuture<VirtualHostNode>, VirtualHostNode>()
                       {
                           @Override
                           public ListenableFuture<VirtualHostNode> call(final VirtualHostNode virtualHostNode)
                                   throws Exception
                           {
                               // permission has already been granted to create the virtual host
                               // disable further access check on other operations, e.g. create exchange
                               Subject.doAs(getSubjectWithAddedSystemRights(),
                                            new PrivilegedAction<Object>()
                                            {
                                                @Override
                                                public Object run()
                                                {
                                                    virtualHostNode.start();
                                                    return null;
                                                }
                                            });
                               return Futures.immediateFuture(virtualHostNode);
                           }
                       });
    }

    @Override
    public long getBytesIn()
    {
        return getDataReceiptStatistics().getTotal();
    }

    @Override
    public long getBytesOut()
    {
        return getDataDeliveryStatistics().getTotal();
    }

    @Override
    public long getMessagesIn()
    {
        return getMessageReceiptStatistics().getTotal();
    }

    @Override
    public long getMessagesOut()
    {
        return getMessageDeliveryStatistics().getTotal();
    }

    @SuppressWarnings("unchecked")
    @Override
    public <C extends ConfiguredObject> ListenableFuture<C> addChildAsync(final Class<C> childClass,
                                                                          final Map<String, Object> attributes)
    {
        if (childClass == VirtualHostNode.class)
        {
            return (ListenableFuture<C>) createVirtualHostNodeAsync(attributes);
        }
        else
        {
            return getObjectFactory().createAsync(childClass, attributes, this);
        }


    }

    @Override
    protected ListenableFuture<Void> beforeClose()
    {
        _brokerLoggersToClose = new ArrayList(getChildren(BrokerLogger.class));
        return super.beforeClose();
    }

    @Override
    protected ListenableFuture<Void> onClose()
    {
        if (_reportingTimer != null)
        {
            _reportingTimer.cancel();
        }

        shutdownHouseKeeping();

        stopPreferenceTaskExecutor();
        closePreferenceStore();

        _eventLogger.message(BrokerMessages.STOPPED());

        try
        {
            for (BrokerLogger<?> logger : _brokerLoggersToClose)
            {
                logger.stopLogging();
            }
        }
        finally
        {
            Runnable task = _parent.getOnContainerCloseTask();
            if(task != null)
            {
                task.run();
            }
        }
        return Futures.immediateFuture(null);

    }

    @Override
    public UserPreferences createUserPreferences(final ConfiguredObject<?> object)
    {
        return new UserPreferencesImpl(_preferenceTaskExecutor, object, _preferenceStore, Collections.<Preference>emptySet());
    }

    private void updateAccessControl()
    {
        if(!isManagementMode())
        {
            List<AccessControlProvider> children = new ArrayList<>(getChildren(AccessControlProvider.class));
            Collections.sort(children, CommonAccessControlProvider.ACCESS_CONTROL_PROVIDER_COMPARATOR);

            List<AccessControl<?>> accessControls = new ArrayList<>(children.size()+1);
            accessControls.add(_systemUserAllowed);
            for(AccessControlProvider prov : children)
            {
                if(prov.getState() == State.ERRORED)
                {
                    accessControls.clear();
                    accessControls.add(AccessControl.ALWAYS_DENIED);
                    break;
                }
                else if(prov.getState() == State.ACTIVE)
                {
                    accessControls.add(prov.getAccessControl());
                }

            }

            ((CompoundAccessControl)_accessControl).setAccessControls(accessControls);

        }
    }

    public AccessControl getAccessControl()
    {
        return _accessControl;
    }

    @Override
    public VirtualHost<?> findVirtualHostByName(String name)
    {
        for (VirtualHostNode<?> virtualHostNode : getChildren(VirtualHostNode.class))
        {
            VirtualHost<?> virtualHost = virtualHostNode.getVirtualHost();
            if (virtualHost != null && virtualHost.getName().equals(name))
            {
                return virtualHost;
            }
        }
        return null;
    }

    @Override
    public VirtualHostNode findDefautVirtualHostNode()
    {
        VirtualHostNode existingDefault = null;
        Collection<VirtualHostNode<?>> virtualHostNodes = new ArrayList<>(getVirtualHostNodes());
        for(VirtualHostNode node : virtualHostNodes)
        {
            if (node.isDefaultVirtualHostNode())
            {
                existingDefault = node;
                break;
            }
        }
        return existingDefault;
    }

    @Override
    public Collection<KeyStore<?>> getKeyStores()
    {
        Collection children = getChildren(KeyStore.class);
        return children;
    }

    @Override
    public Collection<TrustStore<?>> getTrustStores()
    {
        Collection children = getChildren(TrustStore.class);
        return children;
    }

    @Override
    public boolean isManagementMode()
    {
        return _parent.isManagementMode();
    }

    @Override
    public Collection<AccessControlProvider<?>> getAccessControlProviders()
    {
        Collection children = getChildren(AccessControlProvider.class);
        return children;
    }

    @Override
    protected void onExceptionInOpen(RuntimeException e)
    {
        _eventLogger.message(BrokerMessages.FATAL_ERROR(e.getMessage()));
    }

    @Override
    protected void logOperation(final String operation)
    {
        getEventLogger().message(BrokerMessages.OPERATION(operation));
    }

    public void registerMessageDelivered(long messageSize)
    {
        _messagesDelivered.registerEvent(1L);
        _dataDelivered.registerEvent(messageSize);
    }

    public void registerMessageReceived(long messageSize, long timestamp)
    {
        _messagesReceived.registerEvent(1L, timestamp);
        _dataReceived.registerEvent(messageSize, timestamp);
    }

    @Override
    public long getFlowToDiskThreshold()
    {
        return _flowToDiskThreshold;
    }

    @Override
    public long getNumberOfActivePooledBuffers()
    {
        return QpidByteBuffer.getNumberOfActivePooledBuffers();
    }

    @Override
    public long getNumberOfPooledBuffers()
    {
        return QpidByteBuffer.getNumberOfPooledBuffers();
    }

    public StatisticsCounter getMessageReceiptStatistics()
    {
        return _messagesReceived;
    }

    public StatisticsCounter getDataReceiptStatistics()
    {
        return _dataReceived;
    }

    public StatisticsCounter getMessageDeliveryStatistics()
    {
        return _messagesDelivered;
    }

    public StatisticsCounter getDataDeliveryStatistics()
    {
        return _dataDelivered;
    }

    @Override
    public void resetStatistics()
    {
        _messagesDelivered.reset();
        _dataDelivered.reset();
        _messagesReceived.reset();
        _dataReceived.reset();

        for (VirtualHostNode<?> virtualHostNode : getChildren(VirtualHostNode.class))
        {
            VirtualHost<?> virtualHost = virtualHostNode.getVirtualHost();
            if (virtualHost instanceof StatisticsGatherer)
            {
                ((StatisticsGatherer)virtualHost).resetStatistics();
            }
        }
    }

    private class StatisticsReportingTask extends TimerTask
    {
        private final int DELIVERED = 0;
        private final int RECEIVED = 1;

        private final boolean _reset;
        private final EventLogger _logger;
        private final Subject _subject;

        public StatisticsReportingTask(boolean reset, EventLogger logger)
        {
            _reset = reset;
            _logger = logger;
            _subject = getSystemTaskSubject("Statistics");
        }

        public void run()
        {
            Subject.doAs(_subject, new PrivilegedAction<Object>()
            {
                @Override
                public Object run()
                {
                    reportStatistics();
                    return null;
                }
            });
        }

        protected void reportStatistics()
        {
            try
            {
                _eventLogger.message(BrokerMessages.STATS_DATA(DELIVERED, _dataDelivered.getPeak() / 1024.0, _dataDelivered.getTotal()));
                _eventLogger.message(BrokerMessages.STATS_MSGS(DELIVERED, _messagesDelivered.getPeak(), _messagesDelivered.getTotal()));
                _eventLogger.message(BrokerMessages.STATS_DATA(RECEIVED, _dataReceived.getPeak() / 1024.0, _dataReceived.getTotal()));
                _eventLogger.message(BrokerMessages.STATS_MSGS(RECEIVED,
                                                               _messagesReceived.getPeak(),
                                                               _messagesReceived.getTotal()));

                for (VirtualHostNode<?> virtualHostNode : getChildren(VirtualHostNode.class))
                {
                    VirtualHost<?> virtualHost = virtualHostNode.getVirtualHost();
                    if (virtualHost instanceof QueueManagingVirtualHost)
                    {
                        QueueManagingVirtualHost queueVhost = (QueueManagingVirtualHost) virtualHost;
                        String name = virtualHost.getName();
                        StatisticsCounter dataDelivered = queueVhost.getDataDeliveryStatistics();
                        StatisticsCounter messagesDelivered = queueVhost.getMessageDeliveryStatistics();
                        StatisticsCounter dataReceived = queueVhost.getDataReceiptStatistics();
                        StatisticsCounter messagesReceived = queueVhost.getMessageReceiptStatistics();
                        EventLogger logger = queueVhost.getEventLogger();
                        logger.message(VirtualHostMessages.STATS_DATA(name,
                                                                      DELIVERED,
                                                                      dataDelivered.getPeak() / 1024.0,
                                                                      dataDelivered.getTotal()));
                        logger.message(VirtualHostMessages.STATS_MSGS(name,
                                                                      DELIVERED,
                                                                      messagesDelivered.getPeak(),
                                                                      messagesDelivered.getTotal()));
                        logger.message(VirtualHostMessages.STATS_DATA(name,
                                                                      RECEIVED,
                                                                      dataReceived.getPeak() / 1024.0,
                                                                      dataReceived.getTotal()));
                        logger.message(VirtualHostMessages.STATS_MSGS(name,
                                                                      RECEIVED,
                                                                      messagesReceived.getPeak(),
                                                                      messagesReceived.getTotal()));

                    }
                }

                if (_reset)
                {
                    resetStatistics();
                }
            }
            catch(Exception e)
            {
                LOGGER.warn("Unexpected exception occurred while reporting the statistics", e);
            }
        }
    }

    @Override
    public boolean isVirtualHostPropertiesNodeEnabled()
    {
        return _virtualHostPropertiesNodeEnabled;
    }

    public AuthenticationProvider<?> getManagementModeAuthenticationProvider()
    {
        return _managementModeAuthenticationProvider;
    }

    @Override
    public int getNetworkBufferSize()
    {
        return _networkBufferSize;
    }

    public String getDocumentationUrl()
    {
        return _documentationUrl;
    }

    @Override
    public void restart()
    {
        Subject.doAs(getSystemTaskSubject("Broker"), new PrivilegedAction<Void>()
        {
            @Override
            public Void run()
            {
                final SystemConfig<?> systemConfig = (SystemConfig) getParent();
                // This is deliberately asynchronous as the HTTP thread will be interrupted by restarting
                doAfter(systemConfig.setAttributesAsync(Collections.<String,Object>singletonMap(ConfiguredObject.DESIRED_STATE,
                                                                                                State.STOPPED)),
                        new Callable<ListenableFuture<Void>>()
                        {
                            @Override
                            public ListenableFuture<Void> call() throws Exception
                            {
                                return systemConfig.setAttributesAsync(Collections.<String,Object>singletonMap(ConfiguredObject.DESIRED_STATE, State.ACTIVE));
                            }
                        });

                return null;
            }
        });

    }

    @Override
    public Principal getUser()
    {
        return AuthenticatedPrincipal.getCurrentUser();
    }

    @Override
    public SocketConnectionMetaData getConnectionMetaData()
    {
        Subject subject = Subject.getSubject(AccessController.getContext());
        final SocketConnectionPrincipal principal;
        if(subject != null)
        {
            Set<SocketConnectionPrincipal> principals = subject.getPrincipals(SocketConnectionPrincipal.class);
            if(!principals.isEmpty())
            {
                principal = principals.iterator().next();
            }
            else
            {
                principal = null;
            }
        }
        else
        {
            principal = null;
        }
        return principal == null ? null : principal.getConnectionMetaData();
    }

    @Override
    public Set<Principal> getGroups()
    {
        Subject currentSubject = Subject.getSubject(AccessController.getContext());
        if (currentSubject == null)
        {
            return Collections.emptySet();
        }

        final Set<Principal> currentPrincipals = Collections.<Principal>unmodifiableSet(currentSubject.getPrincipals(GroupPrincipal.class));
        return currentPrincipals;
    }

    @Override
    public void purgeUser(final AuthenticationProvider<?> origin, final String username)
    {
        doPurgeUser(origin, username);
    }

    private void doPurgeUser(final AuthenticationProvider<?> origin, final String username)
    {
        // remove from AuthenticationProvider
        if (origin instanceof PasswordCredentialManagingAuthenticationProvider)
        {
            try
            {
                ((PasswordCredentialManagingAuthenticationProvider) origin).deleteUser(username);
            }
            catch (AccountNotFoundException e)
            {
                // pass
            }
        }

        // remove from Groups
        final Collection<GroupProvider> groupProviders = getChildren(GroupProvider.class);
        for (GroupProvider<?> groupProvider : groupProviders)
        {
            final Collection<Group> groups = groupProvider.getChildren(Group.class);
            for (Group<?> group : groups)
            {
                final Collection<GroupMember> members = group.getChildren(GroupMember.class);
                for (GroupMember<?> member : members)
                {
                    if (username.equals(member.getName()))
                    {
                        member.delete();
                    }
                }
            }
        }

        // remove Preferences from all ConfiguredObjects
        Subject userSubject = new Subject(true,
                                          Collections.singleton(new AuthenticatedPrincipal(new UsernamePrincipal(username, origin))),
                                          Collections.EMPTY_SET,
                                          Collections.EMPTY_SET);
        java.util.Queue<ConfiguredObject<?>> configuredObjects = new LinkedList<>();
        configuredObjects.add(BrokerImpl.this);
        while (!configuredObjects.isEmpty())
        {
            final ConfiguredObject<?> currentObject = configuredObjects.poll();
            final Collection<Class<? extends ConfiguredObject>> childClasses = getModel().getChildTypes(currentObject.getClass());
            for (Class<? extends ConfiguredObject> childClass : childClasses)
            {
                final Collection<? extends ConfiguredObject> children = currentObject.getChildren(childClass);
                for (ConfiguredObject child : children)
                {
                    configuredObjects.add(child);
                }
            }

            Subject.doAs(userSubject, new PrivilegedAction<Void>()
            {
                @Override
                public Void run()
                {
                    currentObject.getUserPreferences().delete(null, null, null);
                    return null;
                }
            });
        }
    }

    protected void shutdownHouseKeeping()
    {
        if(_houseKeepingTaskExecutor != null)
        {
            _houseKeepingTaskExecutor.shutdown();

            try
            {
                if (!_houseKeepingTaskExecutor.awaitTermination(HOUSEKEEPING_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS))
                {
                    _houseKeepingTaskExecutor.shutdownNow();
                }
            }
            catch (InterruptedException e)
            {
                LOGGER.warn("Interrupted during Housekeeping shutdown:", e);
                Thread.currentThread().interrupt();
            }
        }
    }

    @Override
    public long getCompactMemoryThreshold()
    {
        return _compactMemoryThreshold;
    }

    @Override
    public long getCompactMemoryInterval()
    {
        return _compactMemoryInterval;
    }

    @Override
    public double getMemoryOccupancyThreshold()
    {
        return _memoryOccupancyThreshold;
    }

    @Override
    public long getMemoryCompactionIncrement()
    {
        return _memoryCompactionIncrement;
    }

    @Override
    public void compactMemory()
    {
        long memOccupiedByMessages = getMemOccupiedByMessages();
        double ratio = memOccupiedByMessages / (double) QpidByteBuffer.getAllocatedDirectMemorySize();

        if (ratio < _memoryOccupancyThreshold)
        {
            int numberOfActivePooledBuffers = QpidByteBuffer.getNumberOfActivePooledBuffers();
            LOGGER.debug("Compacting direct memory buffers: "
                         + "memOccupiedByMessages: {}, numberOfActivePooledBuffers: {}, ratio: {}",
                         memOccupiedByMessages, numberOfActivePooledBuffers, ratio);

            long largestBufferId = QpidByteBuffer.getLargestPooledBufferId();
            _smallestAllowedBufferId = Math.min(largestBufferId,
                                                _smallestAllowedBufferId + _memoryCompactionIncrement);

            final Collection<VirtualHostNode<?>> vhns = getVirtualHostNodes();
            List<ListenableFuture<Void>> futures = new ArrayList<>(vhns.size());
            for (VirtualHostNode<?> vhn : vhns)
            {
                VirtualHost<?> vh = vhn.getVirtualHost();
                if (vh instanceof QueueManagingVirtualHost)
                {
                    ListenableFuture<Void> future = ((QueueManagingVirtualHost) vh).reallocateMessages(
                            _smallestAllowedBufferId);
                    futures.add(future);
                }
            }
            final ListenableFuture<List<Void>> combinedFuture = Futures.allAsList(futures);
            addFutureCallback(combinedFuture, new FutureCallback<List<Void>>()
            {
                @Override
                public void onSuccess(final List<Void> result)
                {
                    if (LOGGER.isDebugEnabled())
                    {
                        long memOccupiedByMessages = getMemOccupiedByMessages();
                        double ratio = memOccupiedByMessages / (double) QpidByteBuffer.getAllocatedDirectMemorySize();
                        LOGGER.debug("After compact direct memory buffers: numberOfActivePooledBuffers: {}, ratio: {}",
                                     QpidByteBuffer.getNumberOfActivePooledBuffers(),
                                     ratio);
                    }
                }

                @Override
                public void onFailure(final Throwable t)
                {
                    LOGGER.warn("Unexpected error during direct memory compaction.", t);
                }
            }, _houseKeepingTaskExecutor);
        }
    }

    private long getMemOccupiedByMessages()
    {
        final Collection<VirtualHostNode<?>> vhns = getVirtualHostNodes();

        long memOccupiedByMessages = 0;
        for (VirtualHostNode<?> vhn : vhns)
        {
            VirtualHost<?> vh = vhn.getVirtualHost();
            if (vh instanceof QueueManagingVirtualHost)
            {
                QueueManagingVirtualHost<?> host = (QueueManagingVirtualHost<?>) vh;
                memOccupiedByMessages += host.getTotalDepthOfQueuesBytesIncludingHeader();
            }
        }
        return memOccupiedByMessages;
    }

    private class AddressSpaceRegistry implements SystemAddressSpaceCreator.AddressSpaceRegistry
    {
        private final ConcurrentMap<String, NamedAddressSpace> _systemAddressSpaces = new ConcurrentHashMap<>();

        @Override
        public void registerAddressSpace(final NamedAddressSpace addressSpace)
        {
            _systemAddressSpaces.put(addressSpace.getName(), addressSpace);
        }

        @Override
        public void removeAddressSpace(final NamedAddressSpace addressSpace)
        {
            _systemAddressSpaces.remove(addressSpace.getName(), addressSpace);
        }

        @Override
        public void removeAddressSpace(final String name)
        {
            _systemAddressSpaces.remove(name);
        }

        @Override
        public NamedAddressSpace getAddressSpace(final String name)
        {
            return name == null ? null : _systemAddressSpaces.get(name);
        }

        @Override
        public Broker<?> getBroker()
        {
            return BrokerImpl.this;
        }
    }


    private final class AccessControlProviderListener extends AbstractConfigurationChangeListener
    {
        private final Set<ConfiguredObject<?>> _bulkChanges = new HashSet<>();


        @Override
        public void childAdded(final ConfiguredObject<?> object, final ConfiguredObject<?> child)
        {
            if(object.getCategoryClass() == Broker.class && child.getCategoryClass() == AccessControlProvider.class)
            {
                child.addChangeListener(this);
                BrokerImpl.this.updateAccessControl();
            }
        }

        @Override
        public void childRemoved(final ConfiguredObject<?> object, final ConfiguredObject<?> child)
        {
            if(object.getCategoryClass() == Broker.class && child.getCategoryClass() == AccessControlProvider.class)
            {
                BrokerImpl.this.updateAccessControl();
            }
        }

        @Override
        public void attributeSet(final ConfiguredObject<?> object,
                                 final String attributeName,
                                 final Object oldAttributeValue,
                                 final Object newAttributeValue)
        {
            if(object.getCategoryClass() == AccessControlProvider.class && !_bulkChanges.contains(object))
            {
                BrokerImpl.this.updateAccessControl();
            }
        }

        @Override
        public void bulkChangeStart(final ConfiguredObject<?> object)
        {
            if(object.getCategoryClass() == AccessControlProvider.class)
            {
                _bulkChanges.add(object);
            }
        }

        @Override
        public void bulkChangeEnd(final ConfiguredObject<?> object)
        {
            if(object.getCategoryClass() == AccessControlProvider.class)
            {
                _bulkChanges.remove(object);
                BrokerImpl.this.updateAccessControl();
            }
        }
    }

}