ConnectedCatalog.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 __network_ConnectedCatalog_h__
#define __network_ConnectedCatalog_h__

namespace fe
{
namespace ext
{

#define FE_CNC_MAX_CLIENTS      1024    //* avoid realloc
#define FE_CNC_IDENTITY_BYTES   5

//* TODO maybe extend configure like ZeroMQ endpoint: "tcp://localhost:7890"

/**************************************************************************//**
    @brief StateCatalog with connected mirroring

    Before starting a connection, several Catalog values may need to be set.
    In the short form, the configure method can set up the connection
    with a single parsable string, such as:

        "localhost:7890 role=client connectionBlock=true"

    The first argument in the address and port.
    For a server, the address is ignored,
    although the convention is to just use "*".

    The remaining arguments are name/value pairs.
    The current list a recognized values are as follows:

        role            "server" or "client"
        address         string address for a client to connect to
        port            integer port number
        ioPriority      usually "normal" or "high" (for Windows only)
        transport       "tcp" or "udp", where appropriate
        connectionBlock "true" or "false", to block flushes before connection

    In the long form,
    the internal catalog settings can be written and read directly.
    The long form keys used in the catalog match the short form names,
    but with a "net:" prefix.

    "net:role" should be set to either "server" or "client".

    "net:address" should be set to the string address
    for a client to connect to.
    This is ignored for a server.

    "net:port" should be set to the integer port number.

    "net:ioPriority" can be set to a string, usually either "normal" or "high".
    The additional values are "idle", "lowest", "low", "highest",
    and "critical".
    Currently, this only has effect when using the stdthread module
    under Windows.

    "net:transport" can be set to "tcp" or "udp",
    if appropriate for the implementation.

    "net:connectionBlock" can be set to "true" or "false",
    whether to block flushes when waiting for a connection.

    @ingroup network
*//***************************************************************************/
class FE_DL_EXPORT ConnectedCatalog:
    public StateCatalog,
    public Initialize<ConnectedCatalog>
{
    public:

                ConnectedCatalog(void);
virtual         ~ConnectedCatalog(void);

        void    initialize(void);

                /// @brief Provide a parsable setup string
virtual Result  configure(String a_line) override;

                /** @brief Initiate connections

                    A server will begin seeking client connections
                    and a client will try to make a connection.
                    This method may return before making any connections. */
virtual Result  start(void) override;

                /** @brief Disconnect and stop initiating connections

                    New connections will no longer be accepted.

                    This method will not return until the connections
                    existing connections are disconnected. */
virtual Result  stop(void) override;

                /** @brief Return only after a connection has been made

                    A derived implementation could return a failure result code,
                    indicating a problem prior to making a connection. */
virtual Result  waitForConnection(void) override;

                /** @brief Check if the catalog wants to connect

                    The started state means the catalog is seeking connections.
                    A started catlog may have connections. */
virtual BWORD   started(void) const override
                {   return m_started.load(std::memory_order_relaxed); }

                /** @brief Check if the catalog has connections

                    A catalog with connections is always started. */
virtual BWORD   connected(void) const override
                {   return m_connected.load(std::memory_order_relaxed); }

                /** @brief Indicate how long the server may remain quiet

                    If the server has a gap between flushes exceeding
                    this time limit, the clients may drop the connection.
                    They may try to reconnect with the current
                    connection settings.
                */
virtual void    setConnectionTimeout(Real a_seconds);

                /** @brief Indicate the end of an atomic set of changes

                    This may provoke a transmission of buffered data.

                    Receiving connections should use this to increment
                    the StateCatalog flush count and
                    make a new Snapshot available.
                */
virtual Result  flush(void) override;

virtual Result  waitForUpdate(I32& a_rFlushCount,I32& a_rSpins,
                    volatile BWORD& a_rKeepWaiting,I32 a_microSleep) override;
virtual Result  lockAfterUpdate(I32& a_rFlushCount,I32& a_rSpins,
                    volatile BWORD& a_rKeepWaiting,I32 a_microSleep) override;

    protected:

virtual Result  preGet(String a_name,String a_property) const override;
virtual Result  postSet(String a_name,String a_property) override;

virtual Result  connect(void);
virtual Result  disconnect(void);
virtual Result  dropConnection(void);

virtual Result  connectAsServer(String a_address,U16 a_port);
virtual Result  connectAsClient(String a_address,U16 a_port);

virtual void    broadcastUpdate(String a_name,String a_property);
virtual void    heartbeat(void);
virtual void    service(void);

virtual void    broadcastSelect(String a_name,String a_property,
                        String a_message,
                        I32 a_includeCount,const String* a_pIncludes,
                        I32 a_excludeCount,const String* a_pExcludes,
                        const U8* a_pRawBytes=NULL,I32 a_byteCount=0)       =0;

virtual BWORD   useBinaryForType(String a_type) const;

        void    broadcastMessage(String a_name,String a_message);

        void    broadcastTick(void);

        void    timerRestart(void);
        BWORD   timerReached(void) const;
        Real    timerSeconds(void) const;

        void    tickerRestart(void);
        BWORD   tickerReached(void) const;
        Real    tickerSeconds(void) const;

        const String&   role(void) const            { return m_role; }

        bool            aborting(void) const
                        {   return m_aborting.load(std::memory_order_relaxed); }

        void            requestDisconnect(void)     { m_disconnecting=true; }
        bool            disconnecting(void) const
                        {   return m_disconnecting.load(
                                    std::memory_order_relaxed); }

        void            stall(void) const;
        void            microPause(I32 a_microSeconds) const;
        void            yield(void) const;
        void            serviceYield(void) const;

        enum Command
        {
            e_unknown=0,
            e_update,
            e_signal,
            e_remove
        };

        void            queueRepeat(Command a_command,
                                String a_name,String a_property,
                                String a_message,String a_source,
                                U8* a_pRawBytes,I32 a_byteCount);
        Result          flushRepeats(BWORD a_signalsOnly);

        void            update(Command a_command, String a_source,
                            String a_key,String a_property,
                            String a_type,String a_text,
                            const U8* a_pRawBytes=NULL,I32 a_byteCount=0);

    class Identity
    {
        public:
                            Identity(void):
                                m_initiated(FALSE)                          {}
            void            reset(void)
                            {
                                memset(m_data,0,
                                        FE_CNC_IDENTITY_BYTES*sizeof(U8));
                                m_text="";
                                m_initiated=FALSE;
                            }
            Identity&       operator=(const Identity& a_rOther)
                            {
                                memcpy(m_data,a_rOther.m_data,
                                        FE_CNC_IDENTITY_BYTES);
                                m_text=a_rOther.m_text;
                                m_initiated=a_rOther.m_initiated;
                                return *this;
                            }
            bool            operator<(const Identity& a_rOther) const
                            {   return (memcmp(m_data,a_rOther.m_data,
                                        FE_CNC_IDENTITY_BYTES)<0); }
            bool            operator==(const Identity& a_rOther) const
                            {   return !memcmp(m_data,a_rOther.m_data,
                                        FE_CNC_IDENTITY_BYTES); }
            bool            operator==(const String& a_rText) const
                            {   return (m_text==a_rText); }

            U8              m_data[FE_CNC_IDENTITY_BYTES];
            String          m_text;
            BWORD           m_initiated;
    };

virtual Result  removeIdentity(I32 a_index);
virtual Result  removeIdentity(const Identity& a_rIdentity);

        Result  catchUpAsNeeded(void);
virtual Result  catchUp(Identity& a_rIdentity);

    class KeyProperty
    {
        public:

        class Bytes: public Counted
        {
            public:
                        Bytes(void):
                            m_pRawBytes(NULL)                               {}
                        ~Bytes(void)
                        {   if(m_pRawBytes) free(m_pRawBytes); }

            U8*     m_pRawBytes;
        };

                    KeyProperty(void):
                        m_command(e_unknown),
                        m_byteCount(0)                                      {}

                    KeyProperty(const KeyProperty& a_keyProperty)
                    {
                        m_command=a_keyProperty.m_command;
                        m_key=a_keyProperty.m_key;
                        m_property=a_keyProperty.m_property;
                        m_message=a_keyProperty.m_message;
                        m_source=a_keyProperty.m_source;
                        m_spBytes=a_keyProperty.m_spBytes;
                        m_byteCount=a_keyProperty.m_byteCount;
                    }

                    KeyProperty(Command a_command,
                        String a_key,String a_property,
                        String a_message="",String a_source="",
                        U8* a_pRawBytes=NULL,I32 a_byteCount=0)
                    {   set(a_command,a_key,a_property,a_message,a_source,
                                a_pRawBytes,a_byteCount); }

            void    set(Command a_command,
                        String a_key,String a_property,
                        String a_message="",String a_source="",
                        U8* a_pRawBytes=NULL,I32 a_byteCount=0)
                    {
                        m_command=a_command;
                        m_key=a_key;
                        m_property=a_property;
                        m_message=a_message;
                        m_source=a_source;
                        m_byteCount=a_byteCount;

                        if(a_pRawBytes)
                        {
                            m_spBytes=new Bytes;
                            m_spBytes->m_pRawBytes=a_pRawBytes;
                        }
                        else
                        {
                            m_spBytes=NULL;
                        }
                    }

            Command     m_command;
            String      m_key;
            String      m_property;
            String      m_message;
            String      m_source;
            sp<Bytes>   m_spBytes;
            I32         m_byteCount;
    };

        Identity            m_identityArray[FE_CNC_MAX_CLIENTS];
        I32                 m_identityCount;

        BWORD               m_lockOnUpdate;

    private:

        void            joinThreads(void);

    class KeyMessage
    {
        public:
                    KeyMessage(void)                                        {}
                    KeyMessage(const KeyMessage& a_keyMessage)
                    {   m_key=a_keyMessage.m_key;
                        m_message=a_keyMessage.m_message; }
                    KeyMessage(String a_key,String a_property):
                        m_key(a_key),
                        m_message(a_property)                               {}
            String  m_key;
            String  m_message;
    };

    class Change
    {
        public:
                    Change(void):
                        m_byteCount(0),
                        m_pRawBytes(NULL)                                   {}
                    Change(const Change& a_change)
                    {   m_command=a_change.m_command;
                        m_source=a_change.m_source;
                        m_key=a_change.m_key;
                        m_property=a_change.m_property;
                        m_type=a_change.m_type;
                        m_text=a_change.m_text;
                        m_byteCount=a_change.m_byteCount;
                        m_pRawBytes=a_change.m_pRawBytes; }
            Command m_command;
            String  m_source;
            String  m_key;
            String  m_property;
            String  m_type;
            String  m_text;
            I32     m_byteCount;
            U8*     m_pRawBytes;
    };

    class FE_DL_EXPORT ConnectionTask: public Thread::Functor
    {
        public:
                            ConnectionTask(void);
    virtual                 ~ConnectionTask(void);
    virtual void            operate(void);

            ConnectedCatalog*   m_pConnectedCatalog;
    };

    class FE_DL_EXPORT ServiceTask: public Thread::Functor
    {
        public:
                            ServiceTask(void);
    virtual                 ~ServiceTask(void);
    virtual void            operate(void);

            ConnectedCatalog*   m_pConnectedCatalog;
    };

        Result              m_connectionResult;
        Thread*             m_pConnectionThread;
        ConnectionTask      m_connectionTask;

        Thread*             m_pServiceThread;
        ServiceTask         m_serviceTask;

        String              m_role;
        String              m_clientID;
        String              m_listenKey;

        std::atomic<int>    m_signalsQueued;
        std::atomic<bool>   m_started;
        std::atomic<bool>   m_connected;
        std::atomic<bool>   m_aborting;
        std::atomic<bool>   m_disconnecting;

        Array<KeyProperty>  m_flushKeys;
        Array<KeyMessage>   m_flushMessages;
        Array<Change>       m_changes;

        std::map<String, Array<KeyProperty> >   m_repeatQueueMap;

        std::chrono::steady_clock::time_point m_timerStart;
        Real                m_timerLimit;

        std::chrono::steady_clock::time_point m_tickerStart;
        Real                m_tickerLimit;

        RecursiveMutex      m_startStopMutex;
};

} /* namespace ext */
} /* namespace fe */

#endif /* __network_ConnectedCatalog_h__ */