ZeroCatalog.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 __zeromq_ZeroCatalog_h__
#define __zeromq_ZeroCatalog_h__

namespace fe
{
namespace ext
{

/**************************************************************************//**
    @brief ConnectedCatalog over ZeroMQ

    @ingroup zeromq
*//***************************************************************************/
class FE_DL_EXPORT ZeroCatalog:
    public ConnectedCatalog,
    public Initialize<ZeroCatalog>
{
    public:

                ZeroCatalog(void);
virtual         ~ZeroCatalog(void);

        void    initialize(void);

    protected:

virtual Result  disconnect(void) override;

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

    private:

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) override;

        void    sendString(String a_textD);
        void    sendBytes(const U8* a_pByteBlock,I32 a_byteCountD);

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

            Result  connectAsServer(String a_transport,
                            String a_address,U16 a_port);
            Result  connectAsClient(String a_transport,
                            String a_address,U16 a_port);
            Result  disconnect(void);

            I32     readBuffer(void** a_ppBuffer,zmq_msg_t* a_pMsg);
            I32     readString(String& a_rString);
            I32     readBytes(Array<U8>& a_rByteArray);
            I32     readIdentity(Identity& a_identity);

            ZeroCatalog*        m_pZeroCatalog;
            String              m_endpoint;
            void*               m_pContext;
            void*               m_pSocket;
            std::atomic<bool>   m_terminate;
    };

    //* NOTE one recipient for N messages (using count)
    class FE_DL_EXPORT Messages: public ObjectSafeShared<Messages>
    {
        public:
                                Messages(void):
                                    m_available(FALSE)                      {}
            Array<Identity>     m_recipientArray;
            Array<I32>          m_subCountArray;
            Array<zmq_msg_t>    m_messageArray;
            std::atomic<bool>   m_available;
    };

        Thread*             m_pReceiverThread;
        ReceiverTask        m_receiverTask;
        Messages            m_messages;
};

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

#endif /* __zeromq_ZeroCatalog_h__ */