Scribe.h

Go to the documentation of this file.

//
// Copyright (C) 2006 Institut fuer Telematik, Universitaet Karlsruhe (TH)
//
// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License
// as published by the Free Software Foundation; either version 2
// of the License, or (at your option) any later version.
//
// This program 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, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
//

#ifndef __SCRIBE_H_
#define __SCRIBE_H_

#include <map>
#include <string>

#include "BaseApp.h"
#include "CommonMessages_m.h"
#include "GlobalNodeList.h"
#include "NodeHandle.h"

#include "ScribeGroup.h"
#include "ScribeMessage_m.h"

// Output function for grouplist, needed for WATCH
std::ostream& operator<< (std::ostream& o, std::map<OverlayKey, ScribeGroup> m )
{
    for (std::map<OverlayKey, ScribeGroup>::iterator it = m.begin(); it != m.end(); ++it) {
        o << it->first << "\n";
        o << "  Parent: " << it->second.getParent() << "\n";
        o << "  Status: " << (it->second.getSubscription() ? "Subscriber\n" : "Forwarder\n");
        o << "  Children (" << it->second.numChildren() << "):\n";
        std::set<NodeHandle>::iterator iit = it->second.getChildrenBegin();
        for (int i = it->second.numChildren(); i > 0; --i) {
            o << "    " << *iit << "\n";
            ++iit;
        }
    }
    return o;
}


class Scribe : public BaseApp
{
    private:
        typedef std::map<OverlayKey, ScribeGroup> GroupList;
        GroupList groupList;
        typedef std::multimap<NodeHandle, ScribeTimer*> ChildTimeoutList;
        ChildTimeoutList childTimeoutList;

        int childTimeout;
        int parentTimeout;

        ScribeTimer* subscriptionTimer;

        // statistics
        int numJoins;
        int numChildTimeout;
        int numParentTimeout;
        int numForward;
        int forwardBytes;
        int numReceived;
        int receivedBytes;
        int numHeartbeat;
        int heartbeatBytes;
        int numSubscriptionRefresh;
        int subscriptionRefreshBytes;

    public:
        Scribe( );
        ~Scribe( );

        // see BaseOverlay.h
        virtual void initializeApp( int stage );

        virtual void handleUpperMessage( cMessage *msg );
        virtual void handleReadyMessage( CompReadyMessage* msg );

        virtual void handleTimerEvent( cMessage *msg );

        virtual bool handleRpcCall( BaseCallMessage* msg );
        virtual void handleRpcResponse( BaseResponseMessage* msg,
                                        cPolymorphic* context,
                                        int rpcId, simtime_t rtt );

        virtual void forward(OverlayKey* key, cPacket** msg,
                NodeHandle* nextHopNode);

        virtual void deliver(OverlayKey& key, cMessage* msg);
        virtual void update( const NodeHandle& node, bool joined );

        virtual void finishApp( );

    protected:
        void handleJoinResponse( ScribeJoinResponse* joinResponse );

        void handleJoinCall( ScribeJoinCall* joinMsg);

        void handlePublishCall( ScribePublishCall* publishCall );

        void handlePublishResponse( ScribePublishResponse* publishResponse );

        void handleJoinMessage( ScribeJoinCall* joinMsg, bool amIRoot);

        void handleLeaveMessage( ScribeLeaveMessage* leaveMsg );

        void subscribeToGroup( const OverlayKey& groupId );

        void leaveGroup( const OverlayKey& group );

        void startTimer( ScribeTimer* timer );

        void addChildToGroup( const NodeHandle& child, ScribeGroup& group );

        void removeChildFromGroup( const NodeHandle& child, ScribeGroup& group );

        void removeChildFromGroup( ScribeTimer* timer );

        void checkGroupEmpty( ScribeGroup& group );

        void refreshChildTimer( NodeHandle& child, OverlayKey& groupId );

        void deliverALMDataToGroup( ScribeDataMessage* dataMsg );

        void deliverALMDataToRoot( ALMMulticastMessage* mcastMsg );
};

#endif