//
// 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.
//

/**
 * @file MyChord.cc
 * @author Markus Mauch, Ingmar Baumgart, Dilum Bandara
 * @version 0.1.1
 * @date 08/01/2010
 */

#include <GlobalStatistics.h>
#include <Comparator.h>
#include <BootstrapList.h>
#include <GlobalParameters.h>
#include <NeighborCache.h>
#include <MyChordFingerTable.h>
#include <MyChordSuccessorList.h>
#include <MyDHTMessage_m.h>
#include "MyChord.h"

namespace oversim {

Define_Module(MyChord);

MyChord::MyChord()
{
    stabilize_timer = fixfingers_timer = join_timer = NULL;
    fingerTable = NULL;
}


void MyChord::initializeOverlay(int stage)
{
    // because of IPAddressResolver, we need to wait until interfaces
    // are registered, address auto-assignment takes place etc.
    if (stage != MIN_STAGE_OVERLAY)
        return;

    if (iterativeLookupConfig.merge == true) {
        throw cRuntimeError("MyChord::initializeOverlay(): "
              "MyChord doesn't work with iterativeLookupConfig.merge = true!");
    }
   
    // MyChord provides KBR services
    kbr = true;

    // fetch some parameters
    useCommonAPIforward = par("useCommonAPIforward");
    successorListSize = par("successorListSize");
    joinRetry = par("joinRetry");
    stabilizeRetry = par("stabilizeRetry");
    joinDelay = par("joinDelay");
    stabilizeDelay = par("stabilizeDelay");
    fixfingersDelay = par("fixfingersDelay");
    checkPredecessorDelay = par("checkPredecessorDelay");
    aggressiveJoinMode = par("aggressiveJoinMode");
    extendedFingerTable = par("extendedFingerTable");
    numFingerCandidates = par("numFingerCandidates");
    proximityRouting = par("proximityRouting");

	findGroupMembers = par("findGroupMembers");
	if(!extendedFingerTable && findGroupMembers)
	{
        throw cRuntimeError("MyChord::initializeOverlay(): "
								"Can't find group members without an extended finger table");
	}

	givePriorityToGroup = par("givePriorityToGroup");
	if(givePriorityToGroup && !findGroupMembers)
	{
        throw cRuntimeError("MyChord::initializeOverlay(): "
								"Can't give priority to group if group members aren't found");
	}

	maxFindGroupMemberHops = par("maxFindGroupMemberHops");
    memorizeFailedSuccessor = par("memorizeFailedSuccessor");
    mergeOptimizationL1 = par("mergeOptimizationL1");	// merging optimizations
    mergeOptimizationL2 = par("mergeOptimizationL2");
    mergeOptimizationL3 = par("mergeOptimizationL3");
    mergeOptimizationL4 = par("mergeOptimizationL4");

    keyLength = OverlayKey::getLength();
    failedSuccessor = TransportAddress::UNSPECIFIED_NODE;
    missingPredecessorStabRequests = 0;

    // statistics
    joinCount = 0;
    stabilizeCount = 0;
    fixfingersCount = 0;
    notifyCount = 0;
    newsuccessorhintCount = 0;
    joinBytesSent = 0;
    stabilizeBytesSent = 0;
    notifyBytesSent = 0;
    fixfingersBytesSent = 0;
    newsuccessorhintBytesSent = 0;

    // find friend modules
    findFriendModules();

    // add some watches
    WATCH(predecessorNode);
    WATCH(thisNode);
    WATCH(bootstrapNode);
    WATCH(joinRetry);
    WATCH(missingPredecessorStabRequests);

    // self-messages
    join_timer = new cMessage("join_timer");
    stabilize_timer = new cMessage("stabilize_timer");
    fixfingers_timer = new cMessage("fixfingers_timer");
    checkPredecessor_timer = new cMessage("checkPredecessor_timer");
}


MyChord::~MyChord()
{
    // destroy self timer messages
    cancelAndDelete(join_timer);
    cancelAndDelete(stabilize_timer);
    cancelAndDelete(fixfingers_timer);
    cancelAndDelete(checkPredecessor_timer);
}


void MyChord::joinOverlay()
{
    changeState(INIT);		//Initialize
    changeState(BOOTSTRAP);	//Boot strap
}


void MyChord::joinForeignPartition(const NodeHandle &node)
{
    Enter_Method_Silent();

    // create a join call and sent to the bootstrap node.
    MyJoinCall *call = new MyJoinCall("JoinCall");
	call->setGroupId(groupId);
    call->setBitLength(JOINCALL_L(call));

    RoutingType routingType = (defaultRoutingType == FULL_RECURSIVE_ROUTING ||
                               defaultRoutingType == RECURSIVE_SOURCE_ROUTING) ?
                              SEMI_RECURSIVE_ROUTING : defaultRoutingType;

    sendRouteRpcCall(OVERLAY_COMP, node, thisNode.getKey(),
                     call, NULL, routingType, joinDelay);
}


void MyChord::changeState(int toState)
{
    // Defines tasks to be executed when a state change occurs.
    switch (toState) {
    case INIT:				//Initialize 
        state = INIT;
        setOverlayReady(false);
        predecessorNode = NodeHandle::UNSPECIFIED_NODE;	// initialize predecessor
		predecessorGroupId = -1;
        initializeFriendModules();	// initialize finger table & successor list
        updateTooltip();
        if (debugOutput) {			// debug message
            EV << "[MyChord::changeState() @ " << thisNode.getAddress()
            << " (" << thisNode.getKey().toString(16) << ")]\n"
            << "    Entered INIT stage"	<< endl;
        }
        getParentModule()->getParentModule()->bubble("Enter INIT state.");
        break;

    case BOOTSTRAP:				//Boot straping
        state = BOOTSTRAP;		// initiate bootstrap process
        cancelEvent(join_timer);
        // workaround: prevent notificationBoard from taking
        // ownership of join_timer message
        take(join_timer);
        scheduleAt(simTime(), join_timer);	//Schedule now

        // debug message
        if (debugOutput) {
            EV << "[MyChord::changeState() @ " << thisNode.getAddress()
            << " (" << thisNode.getKey().toString(16) << ")]\n"
            << "    Entered BOOTSTRAP stage" << endl;
        }
        getParentModule()->getParentModule()->bubble("Enter BOOTSTRAP state.");

        // find a new bootstrap node and enroll to the bootstrap list
        bootstrapNode = bootstrapList->getBootstrapNode();

        if (bootstrapNode.isUnspecified()) // is this the first node?
		{
            // create new cord ring
            assert(predecessorNode.isUnspecified()); //If I have a predecessor
            bootstrapNode = thisNode;
            changeState(READY);			//Go to ready state
            updateTooltip();
        }
        break;

    case READY:				//Ready state
        state = READY;
        setOverlayReady(true);

        // initiate stabilization protocol
        cancelEvent(stabilize_timer);
        scheduleAt(simTime() + stabilizeDelay, stabilize_timer);
        // initiate finger repair protocol
        cancelEvent(fixfingers_timer);
        scheduleAt(simTime() + fixfingersDelay, fixfingers_timer);
        // initiate predecessor check
        cancelEvent(checkPredecessor_timer);
        if (checkPredecessorDelay > 0)
            scheduleAt(simTime() + checkPredecessorDelay, checkPredecessor_timer);

        // debug message
        if (debugOutput) {
            EV << "[MyChord::changeState() @ " << thisNode.getAddress()
            << " (" << thisNode.getKey().toString(16) << ")]\n"
            << "    Entered READY stage"  << endl;
        }
        getParentModule()->getParentModule()->bubble("Enter READY state.");
        break;
    }
}


void MyChord::handleTimerEvent(cMessage* msg)
{
    if (msg == join_timer)					// catch JOIN timer
        handleJoinTimerExpired(msg);
    else if (msg == stabilize_timer)		// catch STABILIZE timer
        handleStabilizeTimerExpired(msg);
    else if (msg == fixfingers_timer)		// catch FIX_FINGERS timer
        handleFixFingersTimerExpired(msg);
    else if (msg == checkPredecessor_timer)	// catch CHECK_PREDECESSOR timer
	{
        cancelEvent(checkPredecessor_timer);
        scheduleAt(simTime() + checkPredecessorDelay, checkPredecessor_timer);
        if (!predecessorNode.isUnspecified()) 
			pingNode(predecessorNode);
    }
    else									// unknown self message
        error("MyChord::handleTimerEvent(): received unknown self message");
}


void MyChord::handleUDPMessage(BaseOverlayMessage* msg)
{
    MyChordMessage* chordMsg = check_and_cast<MyChordMessage*>(msg);
    switch(chordMsg->getCommand()) {
    case NEWSUCCESSORHINT:
        handleNewSuccessorHint(chordMsg);
        break;
    case FINDGROUPMEMBER:				//Find a group member
        handleFindGroupMember(chordMsg);
        break;
    case FINDGROUPMEMBERREPLY:				//Find a group member
        handleFindGroupMemberReply(chordMsg);
        break;
    default:
        error("handleUDPMessage(): Unknown message type!");
        break;
    }

    delete chordMsg;
}


bool MyChord::handleRpcCall(BaseCallMessage* msg)
{
    if (state != READY) {
        EV << "[MyChord::handleRpcCall() @ " << thisNode.getAddress()
           << " (" << thisNode.getKey().toString(16) << ")]\n"
           << "    Received RPC call and state != READY" << endl;
        return false;
    }

    // delegate messages
    RPC_SWITCH_START( msg )
    // RPC_DELEGATE( <messageName>[Call|Response], <methodToCall> )
	    RPC_DELEGATE( MyJoin, rpcJoin );
    	RPC_DELEGATE( MyNotify, rpcNotify );
	    RPC_DELEGATE( MyStabilize, rpcStabilize );
    	RPC_DELEGATE( MyFixfingers, rpcFixfingers );
    RPC_SWITCH_END( )

    return RPC_HANDLED;
}


void MyChord::handleRpcResponse(BaseResponseMessage* msg,
                              cPolymorphic* context, int rpcId,
                              simtime_t rtt)
{
    RPC_SWITCH_START(msg)
    	RPC_ON_RESPONSE( MyJoin ) 
		{
	        handleRpcJoinResponse(_MyJoinResponse);
    	    EV << "[MyChord::handleRpcResponse() @ " << thisNode.getAddress()
        	<< " (" << thisNode.getKey().toString(16) << ")]\n"
	        << "    Received a Join RPC Response: id=" << rpcId << "\n"
    	    << "    msg=" << *_MyJoinResponse << " rtt=" << rtt << endl;
        	break;
	    }
    	RPC_ON_RESPONSE( MyNotify ) 
		{
        	handleRpcNotifyResponse(_MyNotifyResponse);
	        EV << "[MyChord::handleRpcResponse() @ " << thisNode.getAddress()
    	    << " (" << thisNode.getKey().toString(16) << ")]\n"
        	<< "    Received a Notify RPC Response: id=" << rpcId << "\n"
	        << "    msg=" << *_MyNotifyResponse << " rtt=" << rtt << endl;
        	break;
    	}
    	RPC_ON_RESPONSE( MyStabilize ) 
		{
        	handleRpcStabilizeResponse(_MyStabilizeResponse);
	        EV << "[MyChord::handleRpcResponse() @ " << thisNode.getAddress()
    	    << " (" << thisNode.getKey().toString(16) << ")]\n"
        	<< "    Received a Stabilize RPC Response: id=" << rpcId << "\n"
	        << "    msg=" << *_MyStabilizeResponse << " rtt=" << rtt << endl;
        	break;
    	}
	    RPC_ON_RESPONSE( MyFixfingers ) 
		{
	        handleRpcFixfingersResponse(_MyFixfingersResponse, SIMTIME_DBL(rtt));
    	    EV << "[MyChord::handleRpcResponse() @ " << thisNode.getAddress()
        	<< " (" << thisNode.getKey().toString(16) << ")]\n"
	        << "    Received a Fixfingers RPC Response: id=" << rpcId << "\n"
    	    << "    msg=" << *_MyFixfingersResponse << " rtt=" << rtt << endl;
	        break;
    	}
    RPC_SWITCH_END( )
}


void MyChord::handleRpcTimeout(BaseCallMessage* msg,
                             const TransportAddress& dest,
                             cPolymorphic* context, int rpcId,
                             const OverlayKey&)
{
    RPC_SWITCH_START(msg)
	    RPC_ON_CALL( FindNode ) 
		{
        	EV << "[MyChord::handleRpcTimeout() @ " << thisNode.getAddress()
	        << " (" << thisNode.getKey().toString(16) << ")]\n"
    	    << "    FindNode RPC Call timed out: id=" << rpcId << "\n"
        	<< "    msg=" << *_FindNodeCall << endl;
	        break;
    	}
    	RPC_ON_CALL( MyJoin ) 
		{
        	EV << "[MyChord::handleRpcTimeout() @ " << thisNode.getAddress()
	        << " (" << thisNode.getKey().toString(16) << ")]\n"
    	    << "    Join RPC Call timed out: id=" << rpcId << "\n"
        	<< "    msg=" << *_MyJoinCall << endl;
    	    break;
    	}	
	    RPC_ON_CALL( MyNotify ) 
		{
	        EV << "[MyChord::handleRpcTimeout() @ " << thisNode.getAddress()
    	    << " (" << thisNode.getKey().toString(16) << ")]\n"
        	<< "    Notify RPC Call timed out: id=" << rpcId << "\n"
	        << "    msg=" << *_MyNotifyCall << endl;
	        if (!handleFailedNode(dest)) 
				join();
	        break;
    	}
	    RPC_ON_CALL( MyStabilize ) 
		{
	        EV << "[MyChord::handleRpcTimeout() @ " << thisNode.getAddress()
    	    << " (" << thisNode.getKey().toString(16) << ")]\n"
	        << "    Stabilize RPC Call timed out: id=" << rpcId << "\n"
    	    << "    msg=" << *_MyStabilizeCall << endl;
	        if (!handleFailedNode(dest)) 
				join();
	        break;
    	}
	    RPC_ON_CALL( MyFixfingers ) 
		{
	        EV << "[MyChord::handleRpcTimeout() @ " << thisNode.getAddress()
    	    << " (" << thisNode.getKey().toString(16) << ")]\n"
        	<< "    Fixfingers RPC Call timed out: id=" << rpcId << "\n"
        	<< "    msg=" << *_MyFixfingersCall << endl;
	        break;
    	}
    RPC_SWITCH_END( )
}


int MyChord::getMaxNumSiblings()
{
    return successorListSize;	//Copies of keys are stored in successors
}


int MyChord::getMaxNumRedundantNodes()
{
    return extendedFingerTable ? numFingerCandidates : 1;
}


bool MyChord::isSiblingFor(const NodeHandle& node,
                         const OverlayKey& key,
                         int numSiblings,
                         bool* err)
{
    if (key.isUnspecified())
        error("MyChord::isSiblingFor(): key is unspecified!");

    if (state != READY) 
	{
        *err = true;
        return false;
    }

    if (numSiblings > getMaxNumSiblings())
        opp_error("MyChord::isSiblingFor(): numSiblings too big!");

    if (numSiblings == -1)	// set default number of siblings to consider
		numSiblings = getMaxNumSiblings();

    // if this is the first and only node on the ring, it is responsible
    if ((predecessorNode.isUnspecified()) && (node == thisNode)) 
	{
        if (successorList->isEmpty() || (node.getKey() == key)) 
		{
            *err = false;
            return true;
        } 
		else 
		{
            *err = true;	//Something went wrong
            return false;
        }
    }

	//If request is destined to me & key is between me & my predecessor, I'm the successor
    if ((node == thisNode)
         && (key.isBetweenR(predecessorNode.getKey(), thisNode.getKey())))
	{
        *err = false;
        return true;
    }

	//Are any of my successors is a siblng for key
    NodeHandle prevNode = predecessorNode;
    NodeHandle curNode;
    for (int i = -1; i < (int)successorList->getSize();
         i++, prevNode = curNode) 
	{
        if (i < 0) 
            curNode = thisNode;
        else
            curNode = successorList->getSuccessor(i).nodeHandle;

        if (node == curNode) // is message destined for curNode?
		{
            if (key.isBetweenR(prevNode.getKey(), curNode.getKey())) 
			{
				//You were asking only up to the closest numSiblings nodes to the key
                if (numSiblings <= ((int)successorList->getSize() - i)) 
				{
                    *err = false;
                    return true;
                } 
				else //Asking for too faraway
				{
                    *err = true;
                    return false;
                }
            } 
			else 
			{   // the key doesn't directly belong to this node, but
                // the node could be a sibling for this key
                if (numSiblings <= 1) 
				{
                    *err = false;
                    return false;
                } 
				else 
				{   // In Chord we don't know if we belong to the
                    // replicaSet of one of our predecessors
                    *err = true;
                    return false;
                }
            }
        }
    }

    // node is not in our neighborSet
    *err = true;
    return false;
}


bool MyChord::handleFailedNode(const TransportAddress& failed)
{
    Enter_Method_Silent();

	//If predecessor failed
    if (!predecessorNode.isUnspecified() && failed == predecessorNode)
	{
        predecessorNode = NodeHandle::UNSPECIFIED_NODE;
		predecessorGroupId = -1;
	}

    //TODO const reference -> trying to compare unspec NH
	//Get a copy before it will be removed by successorList->handleFailedNode()
    TransportAddress oldSuccessor = successorList->getSuccessor().nodeHandle;

    if (successorList->handleFailedNode(failed))
        updateTooltip();
    // check pointer for koorde
    if (fingerTable != NULL)
        fingerTable->handleFailedNode(failed);

    // if we had a ring consisting of 2 nodes and our successor seems
    // to be dead. Remove also predecessor because the successor
    // and predecessor are the same node
    if ((!predecessorNode.isUnspecified()) && oldSuccessor == predecessorNode)
	{
        predecessorNode = NodeHandle::UNSPECIFIED_NODE;
		predecessorGroupId = -1;
        callUpdate(predecessorNode, false);	//Inform application
    }

    if (failed == oldSuccessor)
	{
        // schedule next stabilization process now
        if (memorizeFailedSuccessor)
            failedSuccessor = oldSuccessor;
        cancelEvent(stabilize_timer);
        scheduleAt(simTime(), stabilize_timer);
    }

    if (state != READY) 
		return true;

    if (successorList->isEmpty()) 
	{
        // lost our last successor - cancel periodic stabilize tasks
        // and wait for rejoin
        cancelEvent(stabilize_timer);
        cancelEvent(fixfingers_timer);
    }

    return !(successorList->isEmpty());
}


NodeVector* MyChord::findNode(const OverlayKey& key,
                            int numRedundantNodes,
                            int numSiblings,
                            BaseOverlayMessage* msg)
{
    bool err;
    NodeVector* nextHop;

    if (state != READY)
        return new NodeVector();

    if (successorList->isEmpty() && !predecessorNode.isUnspecified()) 
	{
        throw new cRuntimeError("MyChord: Node is READY, has a "
                                "predecessor but no successor!");
        join();
        return new NodeVector();
    }
    if (key.isUnspecified()) // if key is unspecified, message is for this node
	{
        nextHop = new NodeVector();
        nextHop->push_back(thisNode);	//Return myself
    }
    else if (isSiblingFor(thisNode, key, 1, &err)) // message is destined for this node
	{
        nextHop = new NodeVector();		
        nextHop->push_back(thisNode);	//Add myself
		//Also add my successors
        for (uint32_t i = 0; i < successorList->getSize(); i++)
            nextHop->push_back(successorList->getSuccessor(i).nodeHandle);

        nextHop->downsizeTo(numSiblings); //Reduce if too much
    }
    else if (key.isBetweenR(thisNode.getKey(),	//message destined for our successor
                            successorList->getSuccessor().nodeHandle.getKey()))
	{
        nextHop = new NodeVector();
        for (uint32_t i = 0; i < successorList->getSize(); i++)
            nextHop->push_back(successorList->getSuccessor(i).nodeHandle);
        nextHop->downsizeTo(numRedundantNodes);
    }
    else// find next hop with finger table and/or successor list
    {
//std::cout << "HERE " << key << endl;
		//Do this only to application layer messages. If I'm using groups
		if (findGroupMembers && (msg->getEncapsulatedPacket() != NULL))
		{
    		BaseOverlayMessage* innerMsg = static_cast<BaseOverlayMessage*>(msg->getEncapsulatedPacket());
			if((innerMsg->getEncapsulatedPacket() != NULL) && (innerMsg->getType() == APPDATA))
			{
            	MyDHTGetMessage* dhtGetMsg = dynamic_cast<MyDHTGetMessage*>(innerMsg->getEncapsulatedPacket());
				if(dhtGetMsg != NULL)
				{
					if(givePriorityToGroup && (dhtGetMsg->getGroupId() != -1))
		    			nextHop = closestPreceedingNode(key, dhtGetMsg->getGroupId(), true);
					else if(givePriorityToGroup)
		    			nextHop = closestPreceedingNode(key, groupId, true);
					else
	    				nextHop = closestPreceedingNode(key, -1, true);
				}
				else
	    			nextHop = closestPreceedingNode(key, -1);
    		}
			else
	    		nextHop = closestPreceedingNode(key, -1);
		}
		else
	    	nextHop = closestPreceedingNode(key, -1);
        nextHop->downsizeTo(numRedundantNodes);
    }

    return nextHop;
}


NodeVector* MyChord::closestPreceedingNode(const OverlayKey& key, int id, bool isDHTGetMsg)
{
    NodeHandle tempHandle = NodeHandle::UNSPECIFIED_NODE;

    // find the closest preceding node to key in the successor list
	// This should be the furthest successor
    for (int j = successorList->getSize() - 1; j >= 0; j--) 
	{
        // Identify predecessor of the key, unless we know a node with an Id = destKey
        if (successorList->getSuccessor(j).nodeHandle.getKey().isBetweenR(thisNode.getKey(), key)) 
		{
            tempHandle = successorList->getSuccessor(j).nodeHandle;
            break;
        }
    }

    if(tempHandle.isUnspecified()) 	//None of my succesors are in the range
	{								//This can't happen
        std::stringstream temp;
        temp << "MyChord::closestPreceedingNode(): Successor list broken "
             << thisNode.getKey() << " " << key;
        throw cRuntimeError(temp.str().c_str());
    }

    NodeVector* nextHop = NULL;

    // find the closest preceding node to key in the finger list
	// This should be the furthest finger unless we need to look at Group ID
    for (int i = fingerTable->getSize() - 1; i >= 0; i--) 
	{
        // return a predecessor of the key, unless we know a node with an Id = destKey
        if (fingerTable->getFinger(i).getKey().isBetweenLR(tempHandle.getKey(), key)) 
		{
            nextHop = new NodeVector();
			//If not an extended finger table, don't care about group Id
            if(!extendedFingerTable || !isDHTGetMsg)
			{
                nextHop->push_back(fingerTable->getFinger(i));

                EV << "[MyChord::closestPreceedingNode() @ " << thisNode.getAddress()
                   << " (" << thisNode.getKey().toString(16) << ")]\n"
                   << "    ClosestPreceedingNode: node " << thisNode
                   << " for key " << key << "\n"
                   << "    finger " << fingerTable->getFinger(i).getKey()
                   << " better than \n"
                   << "    " << tempHandle.getKey() << endl;
                return nextHop;
            } 
			else	//Extended finger table. Consider Group ID
			{
          		MySuccessorEntryVector* entry = fingerTable->getFinger(i, key, id, 
																	givePriorityToGroup, tempHandle.getKey());
				for(unsigned int k = 0; k < entry->size(); k++)
					nextHop->push_back((*entry)[k].nodeHandle);	
                return (nextHop);
			}
        }
    }
    nextHop = new NodeVector();
    EV << "[MyChord::closestPreceedingNode() @ " << thisNode.getAddress()
       << " (" << thisNode.getKey().toString(16) << ")]\n"
       << "    No finger found" << endl;

    // if no finger is found, lookup the rest of the successor list
    for (int i = successorList->getSize() - 1; i >= 0
        && nextHop->size() <= numFingerCandidates ; i--) 
	{
        if (successorList->getSuccessor(i).nodeHandle.getKey().isBetween(thisNode.getKey(), key))
            nextHop->push_back(successorList->getSuccessor(i).nodeHandle);
    }

    if (nextHop->size() != 0) 
        return nextHop;

    // if this is the first and only node on the ring, it is responsible
    if ((predecessorNode.isUnspecified()) &&
        (successorList->getSuccessor().nodeHandle == thisNode)) 
	{
        nextHop->push_back(thisNode);
        return nextHop;
    }

    // if there is still no node found throw an exception
    throw cRuntimeError("Error in MyChord::closestPreceedingNode()!");
    return nextHop;
}


void MyChord::recordOverlaySentStats(BaseOverlayMessage* msg)
{
    BaseOverlayMessage* innerMsg = msg;
    while (innerMsg->getType() != APPDATA &&
           innerMsg->getEncapsulatedPacket() != NULL)
	{	
        innerMsg =
            static_cast<BaseOverlayMessage*>(innerMsg->getEncapsulatedPacket());
    }

    switch (innerMsg->getType()) {
        case OVERLAYSIGNALING: 
		{
            MyChordMessage* chordMsg = dynamic_cast<MyChordMessage*>(innerMsg);
            switch(chordMsg->getCommand()) {
            case NEWSUCCESSORHINT:
                RECORD_STATS(newsuccessorhintCount++;
                             newsuccessorhintBytesSent += msg->getByteLength());
                break;
            }
            break;
        }
        case RPC: 
		{
            if ((dynamic_cast<MyStabilizeCall*>(innerMsg) != NULL) ||
                    (dynamic_cast<MyStabilizeResponse*>(innerMsg) != NULL)) 
			{
                RECORD_STATS(stabilizeCount++; stabilizeBytesSent +=
                             msg->getByteLength());
            } 
			else if ((dynamic_cast<MyNotifyCall*>(innerMsg) != NULL) ||
                    (dynamic_cast<MyNotifyResponse*>(innerMsg) != NULL)) 
			{
                RECORD_STATS(notifyCount++; notifyBytesSent +=
                             msg->getByteLength());
            } 
			else if ((dynamic_cast<MyFixfingersCall*>(innerMsg) != NULL) ||
                    (dynamic_cast<MyFixfingersResponse*>(innerMsg) != NULL)) 
			{
                RECORD_STATS(fixfingersCount++; fixfingersBytesSent +=
                             msg->getByteLength());
            } else if ((dynamic_cast<MyJoinCall*>(innerMsg) != NULL) ||
                    (dynamic_cast<MyJoinResponse*>(innerMsg) != NULL)) 
			{
                RECORD_STATS(joinCount++; joinBytesSent += msg->getByteLength());
            }
            break;
        }
        case APPDATA:
            break;

        default:
            throw cRuntimeError("Unknown message type!");
    }
}


void MyChord::finishOverlay()
{
    // remove this node from the bootstrap list
    bootstrapList->removeBootstrapNode(thisNode);

    simtime_t time = globalStatistics->calcMeasuredLifetime(creationTime);
    if (time < GlobalStatistics::MIN_MEASURED) return;

    globalStatistics->addStdDev("MyChord: Sent JOIN Messages/s",
                                joinCount / time);
    globalStatistics->addStdDev("MyChord: Sent NEWSUCCESSORHINT Messages/s",
                                newsuccessorhintCount / time);
    globalStatistics->addStdDev("MyChord: Sent STABILIZE Messages/s",
                                stabilizeCount / time);
    globalStatistics->addStdDev("MyChord: Sent NOTIFY Messages/s",
                                notifyCount / time);
    globalStatistics->addStdDev("MyChord: Sent FIX_FINGERS Messages/s",
                                fixfingersCount / time);
    globalStatistics->addStdDev("MyChord: Sent JOIN Bytes/s",
                                joinBytesSent / time);
    globalStatistics->addStdDev("MyChord: Sent NEWSUCCESSORHINT Bytes/s",
                                newsuccessorhintBytesSent / time);
    globalStatistics->addStdDev("MyChord: Sent STABILIZE Bytes/s",
                                stabilizeBytesSent / time);
    globalStatistics->addStdDev("MyChord: Sent NOTIFY Bytes/s",
                                notifyBytesSent / time);
    globalStatistics->addStdDev("MyChord: Sent FIX_FINGERS Bytes/s",
                                fixfingersBytesSent / time);
}


void MyChord::handleJoinTimerExpired(cMessage* msg)
{
    // only process timer, if node is not bootstrapped yet
    if (state == READY)
        return;

    // enter state BOOTSTRAP
    if (state != BOOTSTRAP)
        changeState(BOOTSTRAP);	//Get a new bootstrap node

    // change bootstrap node from time to time
    joinRetry--;
    if (joinRetry == 0) 
	{
        joinRetry = par("joinRetry");
        changeState(BOOTSTRAP);	//Get a new node & bootstrap
        return;
    }

    // call JOIN RPC
    MyJoinCall* call = new MyJoinCall("JoinCall");
    call->setBitLength(JOINCALL_L(call));
	call->setGroupId(groupId);
    RoutingType routingType = (defaultRoutingType == FULL_RECURSIVE_ROUTING ||
                               defaultRoutingType == RECURSIVE_SOURCE_ROUTING) ?
                              SEMI_RECURSIVE_ROUTING : defaultRoutingType;

    sendRouteRpcCall(OVERLAY_COMP, bootstrapNode, thisNode.getKey(),
                     call, NULL, routingType, joinDelay);

    // schedule next bootstrap process in the case this one fails
    cancelEvent(join_timer);
    scheduleAt(simTime() + joinDelay, msg);
}


void MyChord::handleStabilizeTimerExpired(cMessage* msg)
{
    if (state != READY)
        return;

    // alternative predecessor check
    if ((checkPredecessorDelay == 0) &&
        (missingPredecessorStabRequests >= stabilizeRetry)) 
	{
        // predecessor node seems to be dead. Remove it from predecessor/successor lists
        //successorList->removeSuccessor(predecessorNode);
        predecessorNode = NodeHandle::UNSPECIFIED_NODE;
		predecessorGroupId = -1;
        missingPredecessorStabRequests = 0;
        updateTooltip();
        callUpdate(predecessorNode, false);	//Tell application
    }

    if (!successorList->isEmpty()) // call STABILIZE RPC
	{
        MyStabilizeCall* call = new MyStabilizeCall("StabilizeCall");
        call->setBitLength(STABILIZECALL_L(call));
        sendUdpRpcCall(successorList->getSuccessor().nodeHandle, call);
        missingPredecessorStabRequests++;
    }

    // check if fingers are still alive and remove unreachable finger nodes
    if (mergeOptimizationL4) 
	{
        OverlayKey offset;
        for (uint32_t nextFinger = 0; nextFinger < thisNode.getKey().getLength();
             nextFinger++) 
		{
            offset = OverlayKey::pow2(nextFinger);

            // send message only for non-trivial fingers
            if (offset > successorList->getSuccessor().nodeHandle.getKey() - thisNode.getKey()) 
			{
                if ((fingerTable->getFinger(nextFinger)).isUnspecified()) 
                    continue;
                else
				{
                    pingNode(fingerTable->getFinger(nextFinger), -1, 0, NULL,
                             NULL, NULL, nextFinger);
                }
            }
        }
    }

    // schedule next stabilization process
    cancelEvent(stabilize_timer);
    scheduleAt(simTime() + stabilizeDelay, msg);
}


void MyChord::handleFixFingersTimerExpired(cMessage* msg)
{
    if ((state != READY) || successorList->isEmpty())	//Not ready or I'm the only one
        return;

    OverlayKey offset, lookupKey;
    for (uint32_t nextFinger = 0; nextFinger < thisNode.getKey().getLength();
         nextFinger++)
	{
        // calculate "n + 2^(i - 1)"
        offset = OverlayKey::pow2(nextFinger);
        lookupKey = thisNode.getKey() + offset;

        // send message only for non-trivial fingers
        if (offset > (successorList->getSuccessor().nodeHandle.getKey() - thisNode.getKey())) 
		{
            // call FIXFINGER RPC
            MyFixfingersCall* call = new MyFixfingersCall("FixfingersCall");
            call->setFinger(nextFinger);
			call->setGroupId(groupId);
            call->setBitLength(FIXFINGERSCALL_L(call));

            sendRouteRpcCall(OVERLAY_COMP, lookupKey, call, NULL,
                             DEFAULT_ROUTING, fixfingersDelay);
        } 
		else 	// delete trivial fingers which are held in successor list
            fingerTable->removeFinger(nextFinger);
    }

    // schedule next finger repair process
    cancelEvent(fixfingers_timer);
    scheduleAt(simTime() + fixfingersDelay, msg);
}


void MyChord::handleNewSuccessorHint(MyChordMessage* chordMsg)
{
    MyNewSuccessorHintMessage* newSuccessorHintMsg =
        check_and_cast<MyNewSuccessorHintMessage*>(chordMsg);

    // fetch the successor's predecessor
    NodeHandle predecessor = newSuccessorHintMsg->getPreNode();
    int predecessorGroupId = newSuccessorHintMsg->getPreGroupId();

    // is the successor's predecessor a new successor for this node?
	// or I'm the onlu node
    if (predecessor.getKey().isBetween(thisNode.getKey(),
                                  successorList->getSuccessor().nodeHandle.getKey())
        || (thisNode.getKey() == successorList->getSuccessor().nodeHandle.getKey())) 
	{
        // add the successor's predecessor to the successor list
        successorList->addSuccessor(predecessor, true, predecessorGroupId);
        updateTooltip();
    }

    // if the successor node reports a new successor, put it into the
    // successor list and start stabilizing
    if (mergeOptimizationL3) 
	{
        if (successorList->getSuccessor().nodeHandle == predecessor) 
		{
            MyStabilizeCall *call = new MyStabilizeCall("StabilizeCall");
            call->setBitLength(STABILIZECALL_L(call));
            sendUdpRpcCall(predecessor, call);
        } 
		else 
		{
            if (successorList->getSuccessor().nodeHandle == 
										newSuccessorHintMsg->getSrcNode()) 
			{
                MyStabilizeCall *call = new MyStabilizeCall("StabilizeCall");
                call->setBitLength(STABILIZECALL_L(call));
                sendUdpRpcCall(predecessor, call);
            }
        }
    }
}

void MyChord::handleFindGroupMember(MyChordMessage* chordMsg)
{
    MyFindGroupMemberMessage* findGroupMsg =
        check_and_cast<MyFindGroupMemberMessage*>(chordMsg);

	NodeVector* members = NULL;
	members = fingerTable->getMembersFromFingerTable(thisNode.getKey(),
									findGroupMsg->getKeyLimit(), findGroupMsg->getGroupId());

	//Chechk for group Id in successor list
	//If found reply
	if(members == NULL)
		members = new NodeVector();
    for (unsigned int i = 0; i < successorList->getSize() ; i++) 
	{
		assert(!successorList->getSuccessor(i).nodeHandle.isUnspecified());
		//Match key & makesure next finger position is not exceeded
		if(( findGroupMsg->getGroupId() == successorList->getSuccessor(i).groupId )
			//&& (findGroupMsg->getKeyLimit() > successorList->getSuccessor(i).nodeHandle.getKey()))
			&& successorList->getSuccessor(i).nodeHandle.getKey().isBetween(thisNode.getKey(), findGroupMsg->getKeyLimit()))
		{
			members->push_back(successorList->getSuccessor(i).nodeHandle);
		}
	}
	
	if(members->size() > 0)
	{
		MyFindGroupMemberReply* replyMsg = new MyFindGroupMemberReply("FINDGROUPMEMBERREPLY");
		replyMsg->setCommand(FINDGROUPMEMBERREPLY);
		replyMsg->setGroupId(findGroupMsg->getGroupId());
		replyMsg->setFinger(findGroupMsg->getFinger());
		replyMsg->setSucNodeArraySize(members->size());
		for(unsigned int k = 0; k < members->size(); k++)
			replyMsg->setSucNode(k, (*members)[k]);
		sendMessageToUDP(findGroupMsg->getOriginalSrcNode(), replyMsg);
		return;
	}

	//If a group member is not found & hop count has not expired, forward to successor
	if((findGroupMsg->getHopCount() < (maxFindGroupMemberHops - 1)) && 
									successorList->getSuccessor().nodeHandle.getKey() < findGroupMsg->getKeyLimit()) 
	{
		MyFindGroupMemberMessage* forwardMsg = 
				new MyFindGroupMemberMessage("FINDGROUPMEMBER");
		forwardMsg->setCommand(FINDGROUPMEMBER);
		forwardMsg->setOriginalSrcNode(findGroupMsg->getOriginalSrcNode());
		forwardMsg->setFinger(findGroupMsg->getFinger());
		forwardMsg->setKeyLimit(findGroupMsg->getKeyLimit());
		forwardMsg->setGroupId(findGroupMsg->getGroupId());
		forwardMsg->setHopCount(findGroupMsg->getHopCount() + 1);	//hops++
		sendMessageToUDP(successorList->getSuccessor().nodeHandle, forwardMsg);
	}
}


void MyChord::handleFindGroupMemberReply(MyChordMessage* chordMsg)
{
    MyFindGroupMemberReply* findGroupReply =
        check_and_cast<MyFindGroupMemberReply*>(chordMsg);
	if(( findGroupReply->getGroupId() != groupId) || (!extendedFingerTable))//Wrong Group ID
		return;

	for(unsigned int i = 0; i < findGroupReply->getSucNodeArraySize(); i++)
	{
    	if (findGroupReply->getSucNode(i).isUnspecified())	//Not given
			continue;
                     
		if (findGroupReply->getSucNode(i) == thisNode)		//Me
			continue;

		MySuccessorEntry entry;
		entry.nodeHandle = findGroupReply->getSucNode(i);
		entry.groupId = findGroupReply->getGroupId();
		fingerTable->setAddtionalFinger(findGroupReply->getFinger(), MAXTIME, entry);
	}
}


void MyChord::rpcJoin(MyJoinCall* joinCall)
{
    NodeHandle requestor = joinCall->getSrcNode();
	int requesterGroupId = joinCall->getGroupId();

    // compile successor list
    MyJoinResponse* joinResponse = new MyJoinResponse("JoinResponse");
    int sucNum = successorList->getSize();

    joinResponse->setSrcGroupId(groupId);
    joinResponse->setSucNum(sucNum);
    joinResponse->setSucNodeArraySize(sucNum);
    joinResponse->setSucGroupIdArraySize(sucNum);
    for (int k = 0; k < sucNum; k++)
	{
        joinResponse->setSucNode(k, successorList->getSuccessor(k).nodeHandle);
        joinResponse->setSucGroupId(k, successorList->getSuccessor(k).groupId);
	}
    // sent our predecessor as hint to the joining node
    if (predecessorNode.isUnspecified() && successorList->isEmpty()) 
	{
        joinResponse->setPreNode(thisNode); // we are the only node in the ring
        joinResponse->setPreGroupId(groupId);
	}
	else
	{
        joinResponse->setPreNode(predecessorNode);
        joinResponse->setPreGroupId(predecessorGroupId);
	}

    joinResponse->setBitLength(JOINRESPONSE_L(joinResponse));
    sendRpcResponse(joinCall, joinResponse);

    if (aggressiveJoinMode) 
	{
        // aggressiveJoinMode differs from standard join operations:
        // 1. set our predecessor pointer to the joining node
        // 2. send our old predecessor as hint in JoinResponse msgs
        // 3. send a NEWSUCCESSORHINT to our old predecessor to update
        //    its successor pointer

        // send NEWSUCCESSORHINT to our old predecessor
        if (!predecessorNode.isUnspecified()) 
		{
            MyNewSuccessorHintMessage* newSuccessorHintMsg =
                new MyNewSuccessorHintMessage("NEWSUCCESSORHINT");
            newSuccessorHintMsg->setCommand(NEWSUCCESSORHINT);
            newSuccessorHintMsg->setSrcNode(thisNode);
            newSuccessorHintMsg->setPreNode(requestor);
            newSuccessorHintMsg->setSrcGroupId(groupId);
            newSuccessorHintMsg->setPreGroupId(requesterGroupId);
            newSuccessorHintMsg->
            setBitLength(NEWSUCCESSORHINT_L(newSuccessorHintMsg));

            sendMessageToUDP(predecessorNode, newSuccessorHintMsg);
        }

        // the requestor is our new predecessor
        if (predecessorNode.isUnspecified() || (predecessorNode != requestor)) 
		{
            NodeHandle oldPredecessor = predecessorNode;
            predecessorNode = requestor;
			predecessorGroupId = requesterGroupId;

            // send update to application if we've got a new predecessor
            if (!oldPredecessor.isUnspecified())
                callUpdate(oldPredecessor, false);	//Old one removed
            callUpdate(predecessorNode, true);	//New one added
        }
    }

    // if we don't have a successor, the requestor is also our new successor
    if (successorList->isEmpty())
        successorList->addSuccessor(requestor, true, requesterGroupId);

    updateTooltip();
}


void MyChord::handleRpcJoinResponse(MyJoinResponse* joinResponse)
{
    // determine the numer of successor nodes to add
    int sucNum = successorListSize - 1; //We need space for our entry

    if (joinResponse->getSucNum() < successorListSize - 1)
        sucNum = joinResponse->getSucNum();

    // add successor getNode(s)
    for (int k = 0; k < sucNum; k++) 
	{
        successorList->addSuccessor(joinResponse->getSucNode(k), 
								true, joinResponse->getSucGroupId(k));
    }

    // the sender of this message is our new successor
    successorList->addSuccessor(joinResponse->getSrcNode(), true,
								joinResponse->getSrcGroupId());

    // in aggressiveJoinMode: use hint in JoinResponse to set our new predecessor
    if (aggressiveJoinMode) 
	{
        // it is possible that the joinResponse doesn't contain a valid
        // predecessor especially when merging two partitions
        if (!joinResponse->getPreNode().isUnspecified()) 
		{
            if (!predecessorNode.isUnspecified())
			{
                // inform the original predecessor about the new predecessor
                if (mergeOptimizationL2) 
				{
                    MyNewSuccessorHintMessage* newSuccessorHintMsg =
                        new MyNewSuccessorHintMessage("NEWSUCCESSORHINT");
                    newSuccessorHintMsg->setCommand(NEWSUCCESSORHINT);
                    newSuccessorHintMsg->setSrcNode(thisNode);
                    newSuccessorHintMsg->setPreNode(joinResponse->getPreNode());
            		newSuccessorHintMsg->setSrcGroupId(groupId);
		            newSuccessorHintMsg->setPreGroupId(joinResponse->getPreGroupId());
                    newSuccessorHintMsg->
                        setBitLength(NEWSUCCESSORHINT_L(newSuccessorHintMsg));

                    sendMessageToUDP(predecessorNode, newSuccessorHintMsg);
                }
            }

            NodeHandle oldPredecessor = predecessorNode;
            predecessorNode = joinResponse->getPreNode();	//New predecessor
            predecessorGroupId = joinResponse->getPreGroupId();	

            if (!oldPredecessor.isUnspecified()
                && !joinResponse->getPreNode().isUnspecified()
                && oldPredecessor != joinResponse->getPreNode()) 
			{
                callUpdate(oldPredecessor, false);	//Old one removed
            }
            callUpdate(predecessorNode, true);	//New one added
        }
    }

    updateTooltip();
    changeState(READY);

    // immediate stabilization protocol
    cancelEvent(stabilize_timer);
    scheduleAt(simTime(), stabilize_timer);

    // immediate finger repair protocol
    cancelEvent(fixfingers_timer);
    scheduleAt(simTime(), fixfingers_timer);
}


void MyChord::rpcStabilize(MyStabilizeCall* call)
{
    // our predecessor seems to be alive
    if (!predecessorNode.isUnspecified() && call->getSrcNode() == predecessorNode)
        missingPredecessorStabRequests = 0;

    // reply with StabilizeResponse message
    MyStabilizeResponse* stabilizeResponse =
        new MyStabilizeResponse("StabilizeResponse");
    stabilizeResponse->setPreNode(predecessorNode);
    stabilizeResponse->setSrcGroupId(groupId);
    stabilizeResponse->setPreGroupId(predecessorGroupId);
    stabilizeResponse->setBitLength(STABILIZERESPONSE_L(stabilizeResponse));

    sendRpcResponse(call, stabilizeResponse);
}


void MyChord::handleRpcStabilizeResponse(MyStabilizeResponse* stabilizeResponse)
{
    if (state != READY)
        return;

    // fetch the successor's predecessor
    const NodeHandle& predecessor = stabilizeResponse->getPreNode();

    // is the successor's predecessor a new successor for this node?
    if ((successorList->isEmpty() ||
         predecessor.getKey().isBetween(thisNode.getKey(),
                             successorList->getSuccessor().nodeHandle.getKey())) &&
        (failedSuccessor.isUnspecified() || failedSuccessor != predecessor)) 
	{
        if (successorList->isEmpty() && predecessor.isUnspecified()) 
		{
            // successor is emptry and the sender of the response has
            // no predecessor => take the sender as new successor
            successorList->addSuccessor(stabilizeResponse->getSrcNode(), true,
								stabilizeResponse->getSrcGroupId());
        } 
		else // add the successor's predecessor to the successor list
		{
            successorList->addSuccessor(predecessor, true, 
				stabilizeResponse->getPreGroupId());
		}
        updateTooltip();
    }

    // compile NOTIFY RPC
    MyNotifyCall* notifyCall = new MyNotifyCall("NotifyCall");
    notifyCall->setBitLength(NOTIFYCALL_L(notifyCall));
    notifyCall->setFailed(failedSuccessor);
	notifyCall->setSrcGroupId(groupId);
    failedSuccessor = TransportAddress::UNSPECIFIED_NODE;
	//Inform to successor
    sendUdpRpcCall(successorList->getSuccessor().nodeHandle, notifyCall);
}


void MyChord::rpcNotify(MyNotifyCall* call)
{
    // our predecessor seems to be alive
    if (!predecessorNode.isUnspecified() && call->getSrcNode() == predecessorNode) 
        missingPredecessorStabRequests = 0;

    bool newPredecessorSet = false;	//I'm I going to use the new one?
    NodeHandle newPredecessor = call->getSrcNode();
    int newPredecessorGroupId = call->getSrcGroupId();

    // is the new predecessor closer than the current one?
    if (predecessorNode.isUnspecified() ||
        newPredecessor.getKey().isBetween(predecessorNode.getKey(), thisNode.getKey()) ||
        (!call->getFailed().isUnspecified() &&
         call->getFailed() == predecessorNode)) 
	{
        if ((predecessorNode.isUnspecified()) || (newPredecessor != predecessorNode)) 
		{
            // set up new predecessor
            NodeHandle oldPredecessor = predecessorNode;
            predecessorNode = newPredecessor;
            predecessorGroupId = newPredecessorGroupId;

            if (successorList->isEmpty())	//We are the only ones
                successorList->addSuccessor(newPredecessor, true, newPredecessorGroupId);

            newPredecessorSet = true;
            updateTooltip();

            // send update to application if we've got a new predecessor
            if (!oldPredecessor.isUnspecified())
                callUpdate(oldPredecessor, false); 	//Old one changed
            callUpdate(predecessorNode, true);		//New one added

            // inform the original predecessor about the new predecessor
            if (mergeOptimizationL1) 
			{
                if (!oldPredecessor.isUnspecified()) 
				{
                    MyNewSuccessorHintMessage *newSuccessorHintMsg =
                        new MyNewSuccessorHintMessage("NEWSUCCESSORHINT");
                    newSuccessorHintMsg->setCommand(NEWSUCCESSORHINT);
                    newSuccessorHintMsg->setSrcNode(thisNode);
                    newSuccessorHintMsg->setPreNode(predecessorNode);
            		newSuccessorHintMsg->setSrcGroupId(groupId);
		            newSuccessorHintMsg->setPreGroupId(predecessorGroupId);
                    newSuccessorHintMsg->
                        setBitLength(NEWSUCCESSORHINT_L(newSuccessorHintMsg));
                    sendMessageToUDP(oldPredecessor, newSuccessorHintMsg);
                }
            }
        }
    }

    // compile NOTIFY response
    MyNotifyResponse* notifyResponse = new MyNotifyResponse("NotifyResponse");

    int sucNum = successorList->getSize();
    notifyResponse->setSucNum(sucNum);
    notifyResponse->setSucNodeArraySize(sucNum);
    notifyResponse->setSucGroupIdArraySize(sucNum);
    notifyResponse->setSrcGroupId(groupId);

    // can't accept the notify sender as predecessor,
    // tell it about my correct predecessor
    if (mergeOptimizationL3) 
	{
        if (!newPredecessorSet && (predecessorNode != newPredecessor)) 
		{
            notifyResponse->setPreNode(predecessorNode);
            notifyResponse->setPreGroupId(predecessorGroupId);
            notifyResponse->setPreNodeSet(false);
        } 
		else
            notifyResponse->setPreNodeSet(true);
    }

    for (int k = 0; k < sucNum; k++)
	{ 
        notifyResponse->setSucNode(k, successorList->getSuccessor(k).nodeHandle);
        notifyResponse->setSucGroupId(k, successorList->getSuccessor(k).groupId);
	}
    notifyResponse->setBitLength(NOTIFYRESPONSE_L(notifyResponse));
    sendRpcResponse(call, notifyResponse);
}


void MyChord::handleRpcNotifyResponse(MyNotifyResponse* notifyResponse)
{
    if (state != READY)
        return;

    if (successorList->getSuccessor().nodeHandle != notifyResponse->getSrcNode()) 
	{
        EV << "[MyChord::handleRpcNotifyResponse() @ " << thisNode.getAddress()
           << " (" << thisNode.getKey().toString(16) << ")]\n"
           << "    The srcNode of the received NotifyResponse is not our "
           << " current successor" << endl;
        return;
    }

    // if the NotifyResponse sender couldn't accept me as predecessor,
    // put its predecessor into the successor list and starts stabilizing
    if (mergeOptimizationL3) 
	{
        if (!notifyResponse->getPreNodeSet()) //Didn't accept me
		{
            MyStabilizeCall *call = new MyStabilizeCall("StabilizeCall");
            call->setBitLength(STABILIZECALL_L(call));

            successorList->addSuccessor(notifyResponse->getPreNode(), true,
										notifyResponse->getPreGroupId());
            if (successorList->getSuccessor().nodeHandle == notifyResponse->getPreNode())
                sendUdpRpcCall(notifyResponse->getPreNode(), call);
            return;
        }
    }

    // replace our successor list by our successor's successor list
    successorList->updateList(notifyResponse);
    updateTooltip();
}


void MyChord::rpcFixfingers(MyFixfingersCall* call)
{
	bool groupMemberFound = false;			//Did I found a group member

    MyFixfingersResponse* fixfingersResponse =
        new MyFixfingersResponse("FixfingersResponse");
    fixfingersResponse->setSucNodeArraySize(1);
    fixfingersResponse->setSucNode(0, thisNode);
    fixfingersResponse->setSucGroupIdArraySize(1);
    fixfingersResponse->setSucGroupId(0, groupId);
	if(call->getGroupId() == groupId)			//In same group
		groupMemberFound = true;

	// If extended finger table also send successors
    if (extendedFingerTable) 
	{
		int arraySize;
        if ((successorList->getSize() + 1) < numFingerCandidates + 1)
        	arraySize = successorList->getSize() + 1;
		else
            arraySize = numFingerCandidates + 1;

        fixfingersResponse->setSucNodeArraySize(arraySize);
        fixfingersResponse->setSucGroupIdArraySize(arraySize);

        for (unsigned int i = 0;
            i < (((successorList->getSize()) < numFingerCandidates)
                 ? (successorList->getSize()) : numFingerCandidates); i++) 
		{
            assert(!successorList->getSuccessor(i).nodeHandle.isUnspecified());
            fixfingersResponse->setSucNode(i + 1,
                                           successorList->getSuccessor(i).nodeHandle);
            fixfingersResponse->setSucGroupId(i + 1,
                                           successorList->getSuccessor(i).groupId);
			if(call->getGroupId() == successorList->getSuccessor(i).groupId)
				groupMemberFound = true;
        }
    }

	//If I didn't found a group member ask from my successor
	if((groupMemberFound == false) && findGroupMembers && extendedFingerTable)
	{
		MyFindGroupMemberMessage* findMemberMsg = 
				new MyFindGroupMemberMessage("FINDGROUPMEMBER");
		findMemberMsg->setCommand(FINDGROUPMEMBER);
		findMemberMsg->setOriginalSrcNode(call->getSrcNode());
		findMemberMsg->setFinger(call->getFinger());
		if(call->getFinger() < (OverlayKey::getLength()- 1))
		{
			OverlayKey offset = OverlayKey::pow2(call->getFinger() + 1);//Limit is next finger position
			if(call->getSrcNode().getKey().isBetween(thisNode.getKey(), call->getSrcNode().getKey() + offset))
				findMemberMsg->setKeyLimit(call->getSrcNode().getKey());	//Put a limit
			else
				findMemberMsg->setKeyLimit(call->getSrcNode().getKey() + offset);	//Put a limit
		}
		else
			findMemberMsg->setKeyLimit(call->getSrcNode().getKey());	//Put a limit

		findMemberMsg->setGroupId(call->getGroupId());
		findMemberMsg->setHopCount(0);
		sendMessageToUDP(successorList->getSuccessor().nodeHandle, findMemberMsg);
	}

	//Send fingers
    fixfingersResponse->setFinger(call->getFinger());
    fixfingersResponse->setBitLength(FIXFINGERSRESPONSE_L(fixfingersResponse));

    sendRpcResponse(call, fixfingersResponse);
}


void MyChord::handleRpcFixfingersResponse(MyFixfingersResponse* fixfingersResponse,
                                        double rtt)
{
    // set new finger pointer#
    if (!extendedFingerTable) 
	{
        fingerTable->setFinger(fixfingersResponse->getFinger(),
                               fixfingersResponse->getSucNode(0));
    } 
	else 
	{
        MySuccessors successors;
        for (unsigned int i = 0; i < fixfingersResponse->getSucNodeArraySize();
             i++) 
		{
            if (fixfingersResponse->getSucNode(i).isUnspecified())
                continue;
            if (fixfingersResponse->getSucNode(i) == thisNode)
                break;

			MySuccessorEntry entry;
			entry.nodeHandle = fixfingersResponse->getSucNode(i);
			entry.groupId = fixfingersResponse->getSucGroupId(i);
            successors.insert(std::make_pair(MAXTIME, entry));
        }

        if (successors.size() == 0) 
            return;

        fingerTable->setFinger(fixfingersResponse->getFinger(), successors);

        //if (proximityRouting || globalParameters->getTopologyAdaptation()) {
        if (proximityRouting) 
		{
            for (unsigned int i = 0;
                 i < fixfingersResponse->getSucNodeArraySize(); i++) 
			{
                if (fixfingersResponse->getSucNode(i).isUnspecified())
                    continue;
                if (fixfingersResponse->getSucNode(i) == thisNode)
                    break;
                Prox prox =
                    neighborCache->getProx(fixfingersResponse->getSucNode(i),
                                           NEIGHBORCACHE_DEFAULT,
                                           fixfingersResponse->getFinger(),
                                           this, NULL);
                if (prox == Prox::PROX_TIMEOUT)	//Don't know 
                    fingerTable->removeFinger(fixfingersResponse->getFinger());
                else if (prox != Prox::PROX_UNKNOWN && prox != Prox::PROX_SELF) 
				{
                    fingerTable->updateFinger(fixfingersResponse->getFinger(),
                                              fixfingersResponse->getSucNode(i),
                                              prox.proximity, fixfingersResponse->getSucGroupId(i));
                }
            }
        }
    }
}


void MyChord::proxCallback(const TransportAddress &node, int rpcId,
                         cPolymorphic *contextPointer, Prox prox)
{
    if (prox == Prox::PROX_TIMEOUT) 
	{
        // call join dependant on return value?
        handleFailedNode(node);
        return;
    }

	//TODO Put the correct Group ID while updating. Currently this doesn't get called
    fingerTable->updateFinger(rpcId, (NodeHandle&)node, prox.proximity, -1);
}


void MyChord::pingResponse(PingResponse* pingResponse, cPolymorphic* context,
                         int rpcId, simtime_t rtt)
{
    EV << "[MyChord::pingResponse() @ " << thisNode.getAddress()
       << " (" << thisNode.getKey().toString(16) << ")]\n"
       << "    Received a Ping RPC Response: id=" << rpcId << "\n"
       << "    msg=" << *pingResponse << " rtt=" << rtt
       << endl;

    if (rpcId != -1)
        fingerTable->updateFinger(rpcId, pingResponse->getSrcNode(), 
								  rtt, pingResponse->getGroupId());
}


void MyChord::pingTimeout(PingCall* pingCall,
                        const TransportAddress& dest,
                        cPolymorphic* context, int rpcId)
{
    EV << "[MyChord::pingTimeout() @ " << thisNode.getAddress()
       << " (" << thisNode.getKey().toString(16) << ")]\n"
       << "    Ping RPC timeout: id=" << rpcId << endl;

    // call join dependant on return value?
    handleFailedNode(dest);
}


void MyChord::findFriendModules()
{
    fingerTable = check_and_cast<MyChordFingerTable*>
                  (getParentModule()->getSubmodule("myFingerTable"));

    successorList = check_and_cast<MyChordSuccessorList*>
                    (getParentModule()->getSubmodule("mySuccessorList"));
}


void MyChord::initializeFriendModules()
{
    // initialize finger table
    fingerTable->initializeTable(thisNode.getKey().getLength(), thisNode, this);

    // initialize successor list
    successorList->initializeList(par("successorListSize"), thisNode, this, groupId);
}


void MyChord::updateTooltip()
{
    if (ev.isGUI()) {
        std::stringstream ttString;

        // show our predecessor and successor in tooltip
        ttString << predecessorNode << endl << thisNode << endl
                 << successorList->getSuccessor().nodeHandle;

        getParentModule()->getParentModule()->getDisplayString().
        setTagArg("tt", 0, ttString.str().c_str());
        getParentModule()->getDisplayString().
        setTagArg("tt", 0, ttString.str().c_str());
        getDisplayString().setTagArg("tt", 0, ttString.str().c_str());

        // draw an arrow to our current successor
        showOverlayNeighborArrow(successorList->getSuccessor().nodeHandle, true,
                                 "m=m,50,0,50,0;ls=red,1");
        showOverlayNeighborArrow(predecessorNode, false,
                                 "m=m,50,100,50,100;ls=green,1");
    }
}

// TODO: The following should be removed, since MyChord doesn't have a simple metric
OverlayKey MyChord::distance(const OverlayKey& x,
                           const OverlayKey& y,
                           bool useAlternative) const
{
    return KeyUniRingMetric().distance(x, y);
}

}; //namespace
