core/src/main/java/flex/messaging/FlexSessionManager.java

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

import flex.management.ManageableComponent;
import flex.messaging.log.LogCategories;

import java.util.Enumeration;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Manages FlexSession instances for a MessageBroker.
 */
public class FlexSessionManager extends ManageableComponent {
    public static final String TYPE = "FlexSessionManager";

    private static final long MILLIS_IN_HOUR = 3600000;

    //--------------------------------------------------------------------------
    //
    // Constructors
    //
    //--------------------------------------------------------------------------

    /**
     * Constructs a <tt>FlexSessionManager</tt> for the passed <tt>MessageBroker</tt>.
     *
     * @param broker The root <tt>MessageBroker</tt> using this <tt>FlexSessionManager</tt>.
     */
    public FlexSessionManager(MessageBroker broker) {
        this(false, broker);
    }

    /**
     * Constructs a <tt>FlexSessionManager</tt> for the passed <tt>MessageBroker</tt> and optionally enables management.
     *
     * @param enableManagement <code>true</code> if the <tt>FlexSessionManager</tt>
     *                         is manageable; otherwise <code>false</code>.
     * @param broker           the message broker
     */
    public FlexSessionManager(boolean enableManagement, MessageBroker broker) {
        super(enableManagement);

        super.setId(TYPE);

        this.broker = broker;

        this.setParent(broker);
    }

    //--------------------------------------------------------------------------
    //
    // Variables
    //
    //--------------------------------------------------------------------------

    /**
     * Instance-level lock.
     */
    private final Object lock = new Object();

    //--------------------------------------------------------------------------
    //
    // Properties
    //
    //--------------------------------------------------------------------------

    //----------------------------------
    //  logCategory
    //----------------------------------

    /**
     * Returns the log category for this component.
     *
     * @return The log category for this component.
     */
    @Override
    protected String getLogCategory() {
        return LogCategories.ENDPOINT_FLEXSESSION;
    }

    //----------------------------------
    //  flexSessionCount
    //----------------------------------

    private int flexSessionCount;

    /**
     * Returns the total count of active FlexSessions.
     *
     * @return The total count of active FlexSessions.
     */
    public int getFlexSessionCount() {
        synchronized (lock) {
            return flexSessionCount;
        }
    }

    //----------------------------------
    //  flexSessionProviders
    //----------------------------------

    private final ConcurrentHashMap<Class<? extends FlexSession>, AbstractFlexSessionProvider> providers = new ConcurrentHashMap<Class<? extends FlexSession>, AbstractFlexSessionProvider>();

    /**
     * Returns the registered <tt>FlexSessionProvider</tt> implementation for the specified <tt>FlexSession</tt> type.
     *
     * @param sessionClass The specific <tt>FlexSession</tt> type to get a provider for.
     * @return The registered <tt>FlexSessionProvider</tt> or <code>null</code> if no provider is registered.
     */
    public AbstractFlexSessionProvider getFlexSessionProvider(Class<? extends FlexSession> sessionClass) {
        return providers.get(sessionClass);
    }

    /**
     * Registers a <tt>FlexSessionProvider</tt> implementation for a specified <tt>FlexSession</tt> type.
     *
     * @param sessionClass The specific <tt>FlexSession</tt> type to register a provider for.
     * @param provider     The corresponding <tt>FlexSessionProvider</tt> to register.
     * @return The previously registered provider, or <code>null</code> if no provider was registered for this session type.
     */
    public AbstractFlexSessionProvider registerFlexSessionProvider(Class<? extends FlexSession> sessionClass, AbstractFlexSessionProvider provider) {
        provider.setFlexSessionManager(this);
        AbstractFlexSessionProvider previousProvider = providers.putIfAbsent(sessionClass, provider);

        if (previousProvider != null) {
            previousProvider.stop();
            previousProvider.setFlexSessionManager(null);
        }

        if (isStarted())
            provider.start();

        return previousProvider;
    }

    /**
     * Unregisters a <tt>FlexSessionProvider</tt> implementation for a specified <tt>FlexSession</tt> type.
     *
     * @param sessionClass The specific <tt>FlexSession</tt> type to unregister a provider for.
     */
    public void unregisterFlexSessionProvider(Class<? extends FlexSession> sessionClass) {
        AbstractFlexSessionProvider provider = providers.remove(sessionClass);
        if (provider != null) {
            provider.stop();
            provider.setFlexSessionManager(null);
        }
    }

    //----------------------------------
    //  flexSessions
    //----------------------------------

    /**
     * Registers a new <tt>FlexSession</tt> with the <tt>FlexSessionManager</tt>.
     *
     * @param session The new <tt>FlexSession</tt>.
     */
    public void registerFlexSession(FlexSession session) {
        synchronized (lock) {
            ++flexSessionCount;
            resetMaxFlexSessionsInCurrentHour(flexSessionCount);
        }
    }

    /**
     * Unregisters an invalidated <tt>FlexSession</tt> from the <tt>FlexSessionManager</tt>.
     *
     * @param session The invalidated <tt>FlexSession</tt>.
     */
    public void unregisterFlexSession(FlexSession session) {
        synchronized (lock) {
            --flexSessionCount;
            resetMaxFlexSessionsInCurrentHour(flexSessionCount);
        }
    }

    //----------------------------------
    //  maxFlexSessionsInCurrentHour
    //----------------------------------

    private int maxSessionCountInCurrentHour;
    private long currentHourStartTimestamp = System.currentTimeMillis();

    public int getMaxFlexSessionsInCurrentHour() {
        synchronized (lock) {
            // Make sure we report the correct value if the system has been idle across an hour transition.
            resetMaxFlexSessionsInCurrentHour(flexSessionCount);

            return maxSessionCountInCurrentHour;
        }
    }

    /* Must be called within a synchronized block. */
    private void resetMaxFlexSessionsInCurrentHour(int currentCount) {
        long offset = (System.currentTimeMillis() - currentHourStartTimestamp) / MILLIS_IN_HOUR;
        if (offset > 0) // Shift to the current hour and reset to the current session count.
        {
            currentHourStartTimestamp += (MILLIS_IN_HOUR * offset);
            maxSessionCountInCurrentHour = currentCount;
        } else if (maxSessionCountInCurrentHour < currentCount) {
            maxSessionCountInCurrentHour = currentCount;
        }
    }

    //----------------------------------
    //  messageBroker
    //----------------------------------

    private final MessageBroker broker;

    /**
     * Returns the <tt>MessageBroker</tt> instance that owns this <tt>FlexSessionManager</tt>.
     *
     * @return The parent <tt>MessageBroker</tt> instance.
     */
    public MessageBroker getMessageBroker() {
        return broker;
    }

    //--------------------------------------------------------------------------
    //
    // Public Methods
    //
    //--------------------------------------------------------------------------

    /**
     * Starts the <tt>FlexSessionManager</tt>.
     * Any registered <tt>FlexSession</tt>s providers are also started.
     */
    @Override
    public void start() {
        if (isStarted())
            return;

        for (AbstractFlexSessionProvider provider : providers.values()) {
            if (!provider.isStarted())
                provider.start();
        }

        super.start();
    }

    /**
     * Stops the <tt>FlexSessionManager</tt>.
     * Any registered <tt>FlexSession</tt> providers are stopped and unregistered.
     */
    @Override
    public void stop() {
        if (!isStarted())
            return;

        super.stop();

        Enumeration<Class<? extends FlexSession>> sessionClasses = providers.keys();
        while (sessionClasses.hasMoreElements()) {
            Class<? extends FlexSession> sessionClass = sessionClasses.nextElement();
            unregisterFlexSessionProvider(sessionClass);
        }
        providers.clear();
    }
}