StateCatalog.h

Go to the documentation of this file.

/*  Copyright (C) 2003-2021 Free Electron Organization
    Any use of this software requires a license.  If a valid license
    was not distributed with this file, visit freeelectron.org. */

/** @file */

#ifndef __plugin_StateCatalog_h__
#define __plugin_StateCatalog_h__

#define FE_STATECATALOG_MESSAGE_PROPERTY    "message"

#if FE_CPLUSPLUS >= 201103L
#define FE_STATECATALOG_CALLBACK
#endif

namespace fe
{

/**************************************************************************//**
    @brief Catalog with extensible mirroring

    This class has some pure virtual methods which must be implemented
    by derived classes, such as relaying the state across networks.

    The get/set methods are templated on type.

    @ingroup plugin
*//***************************************************************************/
class FE_DL_EXPORT StateCatalog:
    public ObjectSafeShared<StateCatalog>,
    public Catalog,
    public CastableAs<StateCatalog>
{
    public:

    /** @brief A locking mechanism to get cohesive state

        While the Atomic persists, the StateCatalog will not update.
        The Atomic should be released as soon as possible.
    */
    class Atomic
    {
        public:
                    /// Lock the StateCatalog
                    Atomic(sp<StateCatalog> a_spStateCatalog):
                        m_spStateCatalog(a_spStateCatalog),
                        m_spinCount(0)
                    {
                        FEASSERT(m_spStateCatalog.isValid());
                        m_spStateCatalog->safeLockShared();
                        m_locked=TRUE;
                        m_serial=m_spStateCatalog->serial();
                        m_flushCount=m_spStateCatalog->flushCount();
                    }
                    /** @brief Wait for a state change
                            and then lock the StateCatalog

                        A previous flush count needs to be given
                        as well as a spin counter.

                        This constructor will not return until the
                        flush count of the StateCatalog exceeds the
                        number sent.
                        The reference value is set to the new flush count.

                        A pause time can be specified in micro seconds.
                        If this value is zero, the spin lock is run
                        at full speed.

                        The wait will stop if a_rKeepWaiting is or
                        becomes zero.
                    */
                    Atomic(sp<StateCatalog> a_spStateCatalog,
                        I32& a_rFlushCount,I32 a_microSleep,
                        volatile BWORD& a_rKeepWaiting):
                        m_spStateCatalog(a_spStateCatalog),
                        m_spinCount(0)
                    {
                        FEASSERT(m_spStateCatalog.isValid());
                        m_locked=successful(m_spStateCatalog->lockAfterUpdate(
                                a_rFlushCount,m_spinCount,
                                a_rKeepWaiting,a_microSleep));
                        m_serial=m_spStateCatalog->serial();
                        m_flushCount=m_spStateCatalog->flushCount();
                    }
                    /** @brief Wait for a state change
                            and then lock the StateCatalog

                        This version will always keep waiting.
                    */
                    Atomic(sp<StateCatalog> a_spStateCatalog,
                        I32& a_rFlushCount,I32 a_microSleep=0):
                        m_spStateCatalog(a_spStateCatalog),
                        m_spinCount(0)
                    {
                        BWORD keepWaiting(TRUE);
                        FEASSERT(m_spStateCatalog.isValid());
                        m_locked=successful(m_spStateCatalog->lockAfterUpdate(
                                a_rFlushCount,m_spinCount,
                                keepWaiting,a_microSleep));
                        m_serial=m_spStateCatalog->serial();
                        m_flushCount=m_spStateCatalog->flushCount();
                    }
                    /// Unlock the StateCatalog
                    ~Atomic(void)
                    {
                        FEASSERT(m_spStateCatalog.isValid());
                        if(m_locked)
                        {
                            m_spStateCatalog->safeUnlockShared();
                        }
                    }

                    /** @brief Get state without additional locking
                        (default property)

                        If the lock failed, this method will fail as well.
                    */
        Result      getTypeName(String a_name,String& a_rTypeName) const
                    {   return getTypeName(a_name,"value",a_rTypeName); }

                    /// @brief Get state without additional locking
        Result      getTypeName(String a_name,
                        String a_property,String& a_rTypeName) const
                    {
                        FEASSERT(m_spStateCatalog.isValid());
                        return m_locked?
                                m_spStateCatalog->getTypeNameUnsafe(
                                a_name,a_property,a_rTypeName):
                                e_notInitialized;
                    }

                    /** @brief Get current state without additional locking

                        If the lock failed, this method will fail as well.
                    */
            template <class T>
            Result  getState(String a_name,T& a_rValue) const
                    {   return getState(a_name,"value",a_rValue); }

                    /** @brief Get current state without additional locking

                        If the lock failed, this method will fail as well.
                    */
            template <class T>
            Result  getState(String a_name,
                        String a_property,T& a_rValue) const
                    {
                        FEASSERT(m_spStateCatalog.isValid());
                        return m_locked?
                                m_spStateCatalog->getStateUnsafe(
                                a_name,a_property,a_rValue):
                                e_notInitialized;
                    }

                    /** @brief Return an indication of how long
                            it took for the state to change

                        A spin count of zero means there was a change
                        already in place before this constructor.
                        Gauging different non-zero values may not be
                        a useful metric.
                    */
            I32     spinCount(void) const   { return m_spinCount; }
                    /** @brief Return TRUE if the lock was successful

                        If the a_keepWaiting argument changed becomes zero,
                        the constructor can return without locking.
                        If so, calls to getState will fail.
                    */
            I32     locked(void)    const   { return m_locked; }

                    /** @brief Get the change count
                            from when the Atomic was created

                        The StateCatalog keeps an serial count of how
                        many state changes have occured.
                    */
            I32     serial(void) const      { return m_serial; }

                    /** @brief Get the count of incoming updates
                            from when the Atomic was created

                        An increment of the flushCount corresponds
                        to one flush() call at the sender.
                        Multiple states can be updated on one flush.
                    */
            I32     flushCount(void) const  { return m_flushCount; }

        private:
            sp<StateCatalog>    m_spStateCatalog;
            BWORD               m_locked;
            I32                 m_spinCount;
            I32                 m_serial;
            I32                 m_flushCount;
    };

    /** @brief A fixed copy of state from a StateCatalog

        The state of a snapshot is frozen and can not be changed.
        Unlike the Atomic, a Snapshot does not lock the StateCatalog.
    */
    class Snapshot:
        public Counted,
        public CastableAs<Snapshot>
    {
        public:
            friend class StateCatalog;

                    Snapshot(void):
                        m_serial(-1),
                        m_flushCount(-1)                                    {}
    virtual         ~Snapshot(void)                                         {}

                    /// @brief Get state without any need for locking
                    /// (default property)
        Result      getTypeName(String a_name,String& a_rTypeName) const
                    {   return getTypeName(a_name,"value",a_rTypeName); }

                    /// @brief Get state without any need for locking
        Result      getTypeName(String a_name,
                        String a_property,String& a_rTypeName) const;

                    /// Get historical state without any need for locking
            template <class T>
            Result  getState(String a_name,T& a_rValue) const
                    {   return getState(a_name,"value",a_rValue); }

                    /// Get historical state without any need for locking
            template <class T>
            Result  getState(String a_name,
                        String a_property,T& a_rValue) const
                    {
                        FEASSERT(m_spSnapshotCatalog.isValid());

                        if(!m_spSnapshotCatalog->cataloged(a_name,a_property))
                        {
                            return e_cannotFind;
                        }

                        try
                        {
                            a_rValue=
                                    m_spSnapshotCatalog->catalogOrException<T>(
                                    a_name,a_property);

                            return e_ok;
                        }
                        catch(const Exception& e)
                        {
                            return e_typeMismatch;
                        }
                    }

                    /** @brief Print all the values in the snapshot

                        @internal
                    */
            void    dump(void) const
                    {   m_spSnapshotCatalog->catalogDump(); }
                    /** @brief Print new state changes since the snapshot

                        @internal
                    */
            void    dumpUpdates(void) const
                    {   m_spUpdateCatalog->catalogDump(); }

                    /** @brief Get the change count
                            from when the snapshot was made

                        The StateCatalog keeps an serial count of how
                        many state changes have occured.
                        Each Snapshot taken stores this serial number,
                        which can be used to compare to other snapshots
                        of the same StateCatalog.
                    */
            I32     serial(void) const      { return m_serial; }

                    /** @brief Get the count of incoming updates
                            from when the snapshot was made

                        An increment of the flushCount corresponds
                        to one flush() call at the sender.
                        Multiple states can be updated on one flush.
                    */
            I32     flushCount(void) const  { return m_flushCount; }

        private:
            I32                 m_serial;
            I32                 m_flushCount;
            sp<Catalog>         m_spSnapshotCatalog;
            sp<Catalog>         m_spUpdateCatalog;
    };

    class ListenerI:
        virtual public Component,
        public CastableAs<ListenerI>
    {
        public:
                        /** @brief called when add as a ListenerI to a
                            StateCatalog, giving that object and the key
                            it is listening to  */
    virtual void        bind(sp<StateCatalog> a_spStateCatalog,
                            String a_name)                                  =0;
                        /** @brief called when a key and property are updated
                            */
    virtual void        notify(String a_name,String a_property)             =0;
    };

                    StateCatalog(void):
                        m_serial(0),
                        m_flushCount(0),
                        m_latestSerial(-1)                                  {}
virtual             ~StateCatalog(void)                                     {}

                    /// @brief Provide a parsable setup string
virtual Result      configure(String a_line)                { return e_ok; }

                    /// @brief Begin any special behaviors of derived classes
virtual Result      start(void)                                             =0;
                    /// @brief Cease any special behaviors of derived classes
virtual Result      stop(void)                                              =0;
                    /// @brief Wait for derived class to begin communication
virtual Result      waitForConnection(void)                 { return e_ok; }
                    /// @brief Return TRUE if the special function has begun
virtual BWORD       started(void) const                     { return FALSE; }
                    /// @brief Return TRUE if communication is
                    /// currently established
virtual BWORD       connected(void) const                   { return FALSE; }
                    /** @brief Indicate how long a connection may remain quiet

                        The default implementation does nothing.
                        This may be used by a derived class
                        to determine when a connection has failed.
                    */
virtual void        setConnectionTimeout(Real a_seconds)                    {}
                    /** @brief Indicate the end of some phase of state changes

                        The default implementation does nothing.
                        This may be used by a derived class
                        to trigger a transmission and/or
                        to delineate a single frame in time.
                    */
virtual Result      flush(void)
                    {   sendNotifications();
                        return e_ok; }

                    /** @brief Wait for a state change

                        @internal

                        This method is generally only used by specialized
                        mechanisms, since the Atomic and Snapshot classes
                        neatly present this functionality.

                        See the Atomic constructor for parameter description.
                    */
virtual Result      waitForUpdate(I32& a_rFlushCount,I32& a_rSpins,
                        volatile BWORD& a_rKeepWaiting,I32 a_microSleep)
                    {   return e_ok; }

                    /** @brief Wait for a state change and then lock the mutex

                        @internal

                        This method is generally only used by specialized
                        mechanisms, since the Atomic and Snapshot classes
                        neatly present this functionality.

                        See the Atomic constructor for parameter description.
                    */
virtual Result      lockAfterUpdate(I32& a_rFlushCount,I32& a_rSpins,
                        volatile BWORD& a_rKeepWaiting,I32 a_microSleep)
                    {   return e_unsupported; }

                    /// @brief Get state with mutex (default property)
        Result      getTypeName(String a_name,String& a_rTypeName) const
                    {   return getTypeName(a_name,"value",a_rTypeName); }

                    /// @brief Get state with mutex
        Result      getTypeName(String a_name,
                        String a_property,String& a_rTypeName) const;

                    /** @brief Returns keys that match the regex pattern

                        Keys are appended to the given array.
                        Clear the array if only want new values.
                    */
        void        getStateKeys(Array<String> &a_keys,
                        String a_pattern=".*") const;

                    /** @brief Returns properties for a key

                        Properties are appended to the given array.
                        Clear the array if only want new values.

                        Order is arbitrary.
                    */
        void        getStateProperties(String a_name,
                            Array<String> &a_properties) const;

                    /// @brief Get state with mutex (default property)
        template <class T>
        Result      getState(String a_name,T& a_rValue) const
                    {   return getState(a_name,"value",a_rValue); }

                    /// @brief Get state with mutex
        template <class T>
        Result      getState(String a_name,String a_property,
                        T& a_rValue) const;

        Result      getState(String a_name,String a_property,
                        String& a_rTypeName,String& a_rValue) const;

                    /// @brief Get state with mutex,
                    /// but without calling the internal
                    /// preGet() and postGet() methods
        template <class T>
        Result      justGetState(String a_name,
                        String a_property,T& a_rValue) const;

                    /// @brief Set state with mutex (default property)
        template <class T>
        Result      setState(String a_name,const T& a_rValue)
                    {   return setState(a_name,"value",a_rValue); }

                    /// @brief Set state with mutex
        template <class T>
        Result      setState(String a_name,String a_property,
                        const T& a_rValue);

                    /// @brief Set state with mutex (serialized)
        Result      setState(String a_name,String a_property,
                        String a_typeName,String a_value);

                    /// @brief Remove a state, with mutex
        Result      removeState(String a_name,String a_property);

                    /// @brief Remove all properties of a state, with mutex
        Result      removeState(String a_name);

                    /// @brief Set state for entries in a Catalog, with mutex
        Result      overlayState(sp<Catalog> a_spOtherCatalog);

                    /// @brief Get state without mutex
        Result      getTypeNameUnsafe(String a_name,String& a_rTypeName) const
                    {   return getTypeNameUnsafe(a_name,"value",a_rTypeName); }

                    /// @brief Get state without mutex (default property)
        Result      getTypeNameUnsafe(String a_name,
                        String a_property,String& a_rTypeName) const;

                    /// @brief Get state without mutex
        template <class T>
        Result      getStateUnsafe(String a_name,T& a_rValue) const
                    {   return getStateUnsafe(a_name,"value",a_rValue); }

                    /// @brief Get state without mutex (default property)
        template <class T>
        Result      getStateUnsafe(String a_name,
                        String a_property,T& a_rValue) const;

                    /** @brief Grab current state into a fixed snapshot

                        Multiple calls to get a snapshot will get the
                        same snapshot until there is a change to the state
                        of the StateCatalog.
                    */
        Result      snapshot(sp<Snapshot>& a_rspSnapshot);

                    /** @brief Increment the serial index,
                        indicating a collective change of state

                        @internal

                        This is generally only called from
                        derived implementations,
                        such as when new data arrives over a connection.
                    */
        void        incrementSerial(void)       { m_serial++; }

                    /** @brief Increment the count of incoming flushes
                        indicating a collective remote change of state

                        @internal

                        This is generally only called from
                        derived implementations,
                        such as when new data arrives over a connection.
                    */
        void        incrementFlushCount(void)   { m_flushCount++; }

                    /// @brief Append value to remote queues
        template <class T>
        Result      sendMessage(String a_name,const T& a_rValue);
                    /// @brief Pop received value from front of queue
        template <class T>
        Result      nextMessage(String a_name,T& a_rValue);

                    /** @brief Add a ListenerI for a specific key

                        The bind method of the Listener is called once
                        as it is added here.

                        The notify method of the Listener is called
                        every time any property of that key is updated.

                        The notification arguments are
                        the key name and property.
                    */
        Result      addListener(sp<ListenerI> a_spListener,String a_name);

#ifdef FE_STATECATALOG_CALLBACK
        typedef std::function<void(String,String)> Callback;

                    /** @brief Add a Callback for a specific key

                        The callback is called
                        every time any property of that key is updated.

                        The Callback is simply a
                        std::function<void(String,String)
                        which can be assigned to an appropriate
                        lambda function.

                        The callback arguments are the key name and property.
                    */
        Result      addCallback(Callback& a_lambda,String a_name);
#endif

    protected:

                    /// @internal
virtual Result      preGet(String a_name,String a_property) const;
                    /// @internal
virtual Result      postGet(String a_name,String a_property) const
                    {   return e_ok; }

                    /// @internal
virtual Result      preSet(String a_name,String a_property)     { return e_ok; }
                    /// @internal
virtual Result      postSet(String a_name,String a_property)    { return e_ok; }

        I32         serial(void) const          { return m_serial; }
        I32         flushCount(void) const      { return m_flushCount; }

        Result      addNotification(String a_name,String a_property);
        Result      sendNotifications(void);
        Result      notify(String a_name,String a_property);

        template <class T>
        Result      updateState(String a_name,String a_property,
                        const T& a_rValue);

        Result      updateState(String a_name,String a_property,
                        String a_typeName,String a_value);

        Result      updateState(String a_name,String a_property,
                        String a_typeName,
                        const U8* a_pRawBytes,I32 a_byteCount);

        Result      clearState(String a_name,String a_property);
    private:

        void        cacheInstanceUpdates(String a_name,String a_property,
                        Instance& a_rInstance);

        volatile I32                    m_serial;
        volatile I32                    m_flushCount;

        AutoHashMap<I32, sp<Snapshot> > m_snapshotMap;  //* keyed on serial
        I32                             m_latestSerial;
        sp<Snapshot>                    m_latestSnapshot;

        AutoHashMap<String, Array< sp<ListenerI> > >    m_listenerMap;

#ifdef FE_STATECATALOG_CALLBACK
        AutoHashMap<String, Array<Callback> >           m_callbackMap;
#endif

    class Notification
    {
        public:
                    Notification(void)                                      {}
                    Notification(String a_name,String a_property):
                        m_name(a_name),
                        m_property(a_property)                              {}
            String  m_name;
            String  m_property;
    };

        Array<Notification>         m_notificationArray;
        RecursiveMutex              m_notificationMutex;
};

inline Result StateCatalog::preGet(String a_name,String a_property) const
{
    if(!cataloged(a_name,a_property))
    {
        return e_cannotFind;
    }
    return e_ok;
}

template<class T>
Result StateCatalog::justGetState(String a_name,String a_property,
    T& a_rValue) const
{
    safeLockShared();
    try
    {
        a_rValue=catalogOrException<T>(a_name,a_property);

        safeUnlockShared();
        return e_ok;
    }
    catch(const Exception& e)
    {
        safeUnlockShared();
        return e_cannotFind;
    }
    catch(...)
    {
        safeUnlockShared();
        throw;
    }
}

template<class T>
Result StateCatalog::getState(String a_name,String a_property,T& a_rValue) const
{
    safeLockShared();
    try
    {
        Result result=preGet(a_name,a_property);
        if(failure(result))
        {
            safeUnlockShared();
            return result;
        }

        a_rValue=catalogOrException<T>(a_name,a_property);

        result=postGet(a_name,a_property);
        if(failure(result))
        {
            safeUnlockShared();
            return result;
        }

        safeUnlockShared();
        return e_ok;
    }
    catch(const Exception& e)
    {
        safeUnlockShared();
        return e_cannotFind;
    }
    catch(...)
    {
        safeUnlockShared();
        throw;
    }
}

template<class T>
Result StateCatalog::updateState(String a_name,String a_property,
    const T& a_rValue)
{
    Instance& rInstance=catalogInstance(a_name,a_property);

    rInstance.create<T>(typeMaster());
    rInstance.cast<T>()=a_rValue;

    cacheInstanceUpdates(a_name,a_property,rInstance);

    return addNotification(a_name,a_property);
}

template<class T>
Result StateCatalog::setState(String a_name,String a_property,
    const T& a_rValue)
{
    safeLock();

    Result result=preSet(a_name,a_property);
    if(failure(result))
    {
        safeUnlock();
        return result;
    }

    updateState(a_name,a_property,a_rValue);

    result=postSet(a_name,a_property);
    if(failure(result))
    {
        safeUnlock();
        return result;
    }

    safeUnlock();

    sendNotifications();

    return e_ok;
}

template<class T>
Result StateCatalog::getStateUnsafe(String a_name,String a_property,
    T& a_rValue) const
{
    try
    {
        Result result=preGet(a_name,a_property);
        if(failure(result))
        {
            return result;
        }

        a_rValue=catalogOrException<T>(a_name,a_property);

        result=postGet(a_name,a_property);
        if(failure(result))
        {
            return result;
        }

        return e_ok;
    }
    catch(const Exception& e)
    {
        return e_cannotFind;
    }
    catch(...)
    {
        throw;
    }
}

template<class T>
Result StateCatalog::sendMessage(String a_name,const T& a_rValue)
{
    safeLock();

    Result result=preSet(a_name,FE_STATECATALOG_MESSAGE_PROPERTY);
    if(failure(result))
    {
        safeUnlock();
        return result;
    }

    updateState(a_name,FE_STATECATALOG_MESSAGE_PROPERTY,a_rValue);

    result=postSet(a_name,FE_STATECATALOG_MESSAGE_PROPERTY);
    if(failure(result))
    {
        safeUnlock();
        return result;
    }

    safeUnlock();

    sendNotifications();

    safeLock();

    catalogRemove(a_name,FE_STATECATALOG_MESSAGE_PROPERTY);

    safeUnlock();

    return e_ok;
}

template<class T>
Result StateCatalog::nextMessage(String a_name,T& a_rValue)
{
    return e_unsupported;
}

//* NOTE using std::deque<T> as template to Catalog calls may add the allocator
template<>
inline Result StateCatalog::nextMessage<String>(String a_name,String& a_rValue)
{
    safeLockShared();

    if(!cataloged(a_name))
    {
        safeUnlockShared();
        return e_cannotFind;
    }

    try
    {
        std::deque<String>& rArrayT=
                catalogOrException< std::deque<String> >(a_name,"value");
        if(rArrayT.size()<1)
        {
            safeUnlockShared();
            return e_readFailed;
        }
    }
    catch(const Exception& e)
    {
        FE_MAYBE_UNUSED(e);
        safeUnlockShared();
        return e_cannotFind;
    }
    catch(...)
    {
        safeUnlockShared();
        throw;
    }

    safeUnlockShared();

    safeLock();

    try
    {
        Result result=preGet(a_name,"value");
        if(failure(result))
        {
            safeUnlock();
            return result;
        }

        std::deque<String>& rArrayT=
                catalogOrException< std::deque<String> >(a_name,"value");

        a_rValue=rArrayT.front();
        rArrayT.pop_front();

        result=postGet(a_name,"value");
        if(failure(result))
        {
            safeUnlock();
            return result;
        }
    }
    catch(const Exception& e)
    {
        FE_MAYBE_UNUSED(e);
        safeUnlock();
        return e_cannotFind;
    }
    catch(...)
    {
        safeUnlock();
        throw;
    }

    safeUnlock();
    return e_ok;
}

} /* namespace fe */

#endif /* __plugin_StateCatalog_h__ */