AMDCallbackCollection.h

Go to the documentation of this file.

/*
 * This file is part of ArmarX.
 *
 * Copyright (C) 2011-2016, High Performance Humanoid Technologies (H2T), Karlsruhe Institute of Technology (KIT), all rights reserved.
 *
 * ArmarX is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 as
 * published by the Free Software Foundation.
 *
 * ArmarX is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 *
* @package    ArmarXCore
* @author     Raphael Grimm ( raphael dot grimm at kit dot edu)
* @date       2016
* @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
*             GNU General Public License
*/
#pragma once
 
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
#include <stdexcept>
 
#include <ArmarXCore/interface/core/UserException.h>
 
namespace armarx
{
    /**
     * @brief This helper-class can be used when you have an ice AMD method.
     * It holds the ice function callbacks or custom fuctions to respond or pass an exception..
     * Furthermore an excpetion will be passed if this class destroys and some clients are still waiting.
     *
     * A usecase for this class would be the following scenario:
     * A server S does a long calculation and some clients wait for the result.
     * The server may want to offer a wait function to the clients.
     * Since the server does not want to block threads for waiting the function should be tagged ["amd"] in the slice definition.
     * If a client calls the function, S receives a function callback to send a response or an exception.
     * The server now can use an AMDCallbackCollection to store all callbacks and send an
     * exception or responde to all or some of the stored callbacks.
     * @see https://doc.zeroc.com/pages/viewpage.action?pageId=14031073
     * \ingroup Distributed
     */
    template <class OnCloseExceptionT = ServerShuttingDown>
    class AMDCallbackCollection
    {
    public:
        using ExceptFunction = std::function<void(const std::exception&)>;
        using RespondFunction = std::function<void()>;
 
        AMDCallbackCollection() = default;
        inline ~AMDCallbackCollection();
        /**
         * @brief Adds an callback and processes it depending on the current mode.
         * @param onRespond The function called to responde.
         * @param onExcept The function called on exception.
         */
        inline void addCallback(RespondFunction onRespond, ExceptFunction onExcept);
 
        /**
         * @brief Adds an callback and processes it depending on the current mode.
         * The callback is created using the ice_response and ice_exception functions of the callback object.
         * @param cb The function callback from ice.
         */
        template <class IceCallbackT>
        inline void addCallback(IceUtil::Handle<IceCallbackT> cb);
 
        /**
         * @brief setAutoResponse
         * @param doAutoResponse If set the mode will change to \ref Respond.
         * If false the mode will change from \ref Respond to \ref Store
         */
        inline void setAutoResponse(bool doAutoResponse = true);
 
        /**
         * @brief The n oldest callbacks will receive a response.
         * @param n
         *
         * This may be used in an producer/consumer scenario where the producer creates multiple objects.
         */
        inline void respondeN(std::size_t n);
 
        /**
         * @brief All currently stored callbacks will receive a response.
         */
        inline void respondeAll();
 
        /**
         * @brief If except is null, the mode will be set to \ref Store if it was \ref Except.
         * If except is not null, all currently stored and future callbacks will receive the exception.
         * @param except
         */
        inline void setAutoThrow(std::unique_ptr<std::exception> except);
 
        /**
         * @brief All currently stored callbacks will receive the exception.
         * @param except The exception.
         */
        inline void throwAll(const std::exception& except);
 
        /**
         * @brief All future callbacks will be stored.
         */
        inline void storeFurtherCallbacks();
 
    private:
        struct CallbackHolder
        {
            RespondFunction onRespond;
            ExceptFunction onExcept;
        };
        /**
         * @brief The mode of the class AMDCallbackCollection.
         */
        enum class Mode
        {
            /**
             * @brief All callbacks will be stored.
             */
            Store,
            /**
             * @brief All callbacks will get a response.
             */
            Respond,
            /**
             * @brief All callbacks will get an exception.
             */
            Except
        };
 
        std::deque<CallbackHolder> callbacks;
        Mode currentMode{Mode::Store};
        std::unique_ptr<std::exception> exception;
        std::recursive_mutex classMutex;
    };
 
    template <class OnCloseExceptionT>
    AMDCallbackCollection<OnCloseExceptionT>::~AMDCallbackCollection()
    {
        throwAll(OnCloseExceptionT{"AMDCallbackCollection dtor."});
    }
 
    template <class OnCloseExceptionT>
    void
    AMDCallbackCollection<OnCloseExceptionT>::addCallback(
        AMDCallbackCollection::RespondFunction onRespond,
        AMDCallbackCollection::ExceptFunction onExcept)
    {
        std::lock_guard<std::recursive_mutex> guard{classMutex};
        switch (currentMode)
        {
            case Mode::Except:
                onExcept(*exception);
                return;
            case Mode::Respond:
                onRespond();
                return;
            case Mode::Store:
                callbacks.emplace_back(CallbackHolder{std::move(onRespond), std::move(onExcept)});
                return;
        }
    }
 
    template <class OnCloseExceptionT>
    void
    AMDCallbackCollection<OnCloseExceptionT>::setAutoResponse(bool doAutoResponse)
    {
        std::lock_guard<std::recursive_mutex> guard{classMutex};
        if (doAutoResponse)
        {
            //set mode
            currentMode = Mode::Respond;
            exception = nullptr;
            respondeAll();
            return;
        }
        if (currentMode == Mode::Respond)
        {
            storeFurtherCallbacks();
        }
    }
 
    template <class OnCloseExceptionT>
    void
    AMDCallbackCollection<OnCloseExceptionT>::respondeN(std::size_t n)
    {
        std::lock_guard<std::recursive_mutex> guard{classMutex};
        for (std::size_t i = 0; i < n && !callbacks.empty(); ++i)
        {
            callbacks.at(0).onRespond();
            callbacks.pop_front();
        }
    }
 
    template <class OnCloseExceptionT>
    void
    AMDCallbackCollection<OnCloseExceptionT>::respondeAll()
    {
        std::lock_guard<std::recursive_mutex> guard{classMutex};
        for (auto& cb : callbacks)
        {
            cb.onRespond();
        }
        callbacks.clear();
    }
 
    template <class OnCloseExceptionT>
    void
    AMDCallbackCollection<OnCloseExceptionT>::setAutoThrow(std::unique_ptr<std::exception> except)
    {
        std::lock_guard<std::recursive_mutex> guard{classMutex};
        exception = except;
        if (exception)
        {
            //set exception mode
            currentMode = Mode::Except;
            throwAll(*exception);
            return;
        }
        if (currentMode == Mode::Except)
        {
            //reset exception mode.
            storeFurtherCallbacks();
        }
    }
 
    template <class OnCloseExceptionT>
    void
    AMDCallbackCollection<OnCloseExceptionT>::throwAll(const std::exception& except)
    {
        std::lock_guard<std::recursive_mutex> guard{classMutex};
        for (auto& cb : callbacks)
        {
            cb.onExcept(except);
        }
        callbacks.clear();
    }
 
    template <class OnCloseExceptionT>
    void
    AMDCallbackCollection<OnCloseExceptionT>::storeFurtherCallbacks()
    {
        std::lock_guard<std::recursive_mutex> guard{classMutex};
        exception = nullptr;
        currentMode = Mode::Store;
    }
 
    template <class OnCloseExceptionT>
    template <class IceCallbackT>
    void
    AMDCallbackCollection<OnCloseExceptionT>::addCallback(IceUtil::Handle<IceCallbackT> cb)
    {
        addCallback([cb] { cb->ice_response(); },
                    [cb](const std::exception& e) { cb->ice_exception(e); });
    }
} // namespace armarx