localNode.h

/* Copyright (c) 2005-2013, Stefan Eilemann <eile@equalizergraphics.com>
 *                    2010, Cedric Stalder <cedric.stalder@gmail.com>
 *                    2012, Daniel Nachbaur <danielnachbaur@gmail.com>
 *
 * This file is part of Collage <https://github.com/Eyescale/Collage>
 *
 * This library is free software; you can redistribute it and/or modify it under
 * the terms of the GNU Lesser General Public License version 2.1 as published
 * by the Free Software Foundation.
 *
 * This library 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 Lesser General Public License for more
 * details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#ifndef CO_LOCALNODE_H
#define CO_LOCALNODE_H

#include <co/node.h>            // base class
#include <co/objectHandler.h>   // base class
#include <co/objectVersion.h>   // VERSION_FOO used inline
#include <lunchbox/requestHandler.h> // base class

#include <boost/function/function1.hpp>
#include <boost/function/function4.hpp>

namespace co
{
namespace detail { class LocalNode; class ReceiverThread; class CommandThread; }

    class LocalNode : public lunchbox::RequestHandler, public Node,
                      public ObjectHandler
    {
    public:
        CO_API LocalNode( const uint32_t type = co::NODETYPE_NODE );

        CO_API virtual ~LocalNode();

        CO_API virtual bool initLocal( const int argc, char** argv );

        CO_API virtual bool listen();
        CO_API virtual bool listen( ConnectionPtr connection ); 

        CO_API virtual bool close();

        virtual bool exitLocal() { return close(); }

        CO_API bool connect( NodePtr node );

        CO_API NodePtr connect( const NodeID& nodeID );

        CO_API virtual bool disconnect( NodePtr node );

        CO_API virtual bool registerObject( Object* object );

        CO_API virtual void deregisterObject( Object* object );

        CO_API bool mapObject( Object* object, const UUID& id,
                               const uint128_t& version = VERSION_OLDEST );

        bool mapObject( Object* object, const ObjectVersion& v )
            { return mapObject( object, v.identifier, v.version ); }

        CO_API uint32_t mapObjectNB( Object* object, const UUID& id,
                                     const uint128_t& version = VERSION_OLDEST);

        CO_API virtual uint32_t mapObjectNB( Object* object, const UUID& id,
                                             const uint128_t& version,
                                             NodePtr master );

        CO_API virtual bool mapObjectSync( const uint32_t requestID );

        CO_API virtual void unmapObject( Object* object );

        CO_API void disableInstanceCache();

        CO_API void expireInstanceData( const int64_t age );

        CO_API void enableSendOnRegister();

        CO_API void disableSendOnRegister();

        CO_API virtual void objectPush( const uint128_t& groupID,
                                        const uint128_t& objectType,
                                        const uint128_t& objectID,
                                        DataIStream& istream );

        typedef boost::function< void( const uint128_t&, 
                                       const uint128_t&, 
                                       const uint128_t&, 
                                       DataIStream& ) > PushHandler;
        CO_API void registerPushHandler( const uint128_t& groupID,
                                         const PushHandler& handler );


        typedef boost::function< bool( CustomICommand& ) > CommandHandler;

        CO_API bool registerCommandHandler( const uint128_t& command,
                                            const CommandHandler& func,
                                            CommandQueue* queue );

        CO_API void swapObject( Object* oldObject, Object* newObject );

        CO_API NodePtr getNode( const NodeID& id ) const;

        CO_API void getNodes( Nodes& nodes, const bool addSelf = true ) const;

        CO_API CommandQueue* getCommandThreadQueue();

        CO_API bool inCommandThread() const;

        CO_API int64_t getTime64() const;

        CO_API ConnectionPtr addListener( ConnectionDescriptionPtr desc );

        CO_API void addListener( ConnectionPtr connection );

        CO_API void removeListeners( const Connections& connections );

        CO_API void flushCommands();

        CO_API BufferPtr allocBuffer( const uint64_t size );

        CO_API virtual bool dispatchCommand( ICommand& command );


        typedef lunchbox::RefPtr< co::SendToken > SendToken;

        CO_API SendToken acquireSendToken( NodePtr toNode );

        CO_API void releaseSendToken( SendToken token );

        CO_API Zeroconf getZeroconf();

        CO_API void ackRequest( NodePtr node, const uint32_t requestID );

        CO_API void ping( NodePtr remoteNode );

        CO_API bool pingIdleNodes();

        CO_API void setAffinity( const int32_t affinity );

    protected:
        CO_API bool connect( NodePtr node, ConnectionPtr connection );

        virtual void notifyConnect( NodePtr node ) {}

        virtual void notifyDisconnect( NodePtr node ) {}

        CO_API virtual NodePtr createNode( const uint32_t type );

    private:
        detail::LocalNode* const _impl;

        bool _connectSelf();

        bool _startCommandThread();
        bool _notifyCommandThreadIdle();
        friend class detail::ReceiverThread;
        friend class detail::CommandThread;

        void _cleanup();
        void _closeNode( NodePtr node );
        CO_API void _addConnection( ConnectionPtr connection );
        void _removeConnection( ConnectionPtr connection );

        NodePtr _connect( const NodeID& nodeID, NodePtr peer );
        NodePtr _connectFromZeroconf( const NodeID& nodeID );
        uint32_t _removeListenerNB( ConnectionPtr connection );
        uint32_t _connect( NodePtr node );
        uint32_t _connect( NodePtr node, ConnectionPtr connection );

        void _runReceiverThread();
        void   _handleConnect();
        void   _handleDisconnect();
        bool   _handleData();
        BufferPtr _readHead( ConnectionPtr connection );
        ICommand   _setupCommand( ConnectionPtr, ConstBufferPtr );
        bool      _readTail( ICommand&, BufferPtr, ConnectionPtr );
        void   _initService();
        void   _exitService();

        friend class ObjectStore;
        template< typename T > void
        _registerCommand( const uint32_t command, const CommandFunc< T >& func,
                          CommandQueue* destinationQueue )
        {
            registerCommand( command, func, destinationQueue );
        }

        void _dispatchCommand( ICommand& command );
        void   _redispatchCommands();

        bool _cmdAckRequest( ICommand& command );
        bool _cmdStopRcv( ICommand& command );
        bool _cmdStopCmd( ICommand& command );
        bool _cmdSetAffinity( ICommand& command );
        bool _cmdConnect( ICommand& command );
        bool _cmdConnectReply( ICommand& command );
        bool _cmdConnectAck( ICommand& command );
        bool _cmdID( ICommand& command );
        bool _cmdDisconnect( ICommand& command );
        bool _cmdGetNodeData( ICommand& command );
        bool _cmdGetNodeDataReply( ICommand& command );
        bool _cmdAcquireSendToken( ICommand& command );
        bool _cmdAcquireSendTokenReply( ICommand& command );
        bool _cmdReleaseSendToken( ICommand& command );
        bool _cmdAddListener( ICommand& command );
        bool _cmdRemoveListener( ICommand& command );
        bool _cmdPing( ICommand& command );
        bool _cmdCommand( ICommand& command );
        bool _cmdCommandAsync( ICommand& command );
        bool _cmdAddConnection( ICommand& command );
        bool _cmdDiscard( ICommand& ) { return true; }

        LB_TS_VAR( _cmdThread );
        LB_TS_VAR( _rcvThread );
    };

    inline std::ostream& operator << ( std::ostream& os, const LocalNode& node )
    {
        os << static_cast< const Node& >( node );
        return os;
    }
}
#endif // CO_LOCALNODE_H