node.java

source code for fishnet

import java.util.ArrayList;
import java.util.Iterator;
import java.io.PrintStream;
import java.lang.reflect.Method;

/**
 * <pre>   
 * Node -- Class defining the data structures and code for the
 * protocol stack for a node participating in the fishnet

 * This code is under student control; we provide a baseline
 * "hello world" example just to show how it interfaces with the FishNet classes.
 *
 * NOTE: per-node global variables are prohibited.
 *  In simulation mode, we execute many nodes in parallel, and thus
 *  they would share any globals.  Please put per-node state in class Node
 *  instead.
 *
 * In other words, all the code defined here should be generic -- it shouldn't
 * matter to this code whether it is executing in a simulation or emulation
 * 
 * This code must be written as a state machine -- each upcall must do its work and return so that 
 * other upcalls can be delivered
 * </pre>   
 */
public class Node {
    private final long PingTimeout = 10000;  // Timeout pings in 10 seconds
    
    private Manager manager;
    private int addr;    
    private ArrayList pings; // To store PingRequests.
    

    /**
     * Create a new node
     * @param manager The manager that is managing Fishnet
     * @param addr The address of this node
     */
    public Node(Manager manager, int addr) {
	this.manager = manager;
	this.addr = addr;
	this.pings = new ArrayList();
    }

    /**
     * Called by the manager to start this node up.
     */
    public void start() {
	logOutput("started");
	this.addTimer(PingTimeout, "pingTimedOut");
    }

    /**
     * Called by the manager when a packet has arrived for this node
     * @param from The address of the node that has sent this packet
     * @param msg The serialized form of the packet.
     */
    public void onReceive(Integer from, byte[] msg) {	
	Packet packet = Packet.unpack(msg);
	logOutput("received packet from " + from);
	if(packet == null) {
	    logError("Unable to unpack message: " + Utility.byteArrayToString(msg) + " Received from " + from);
	    return;
	}

	this.receivePacket(from.intValue(), packet);
    }

    /**
     * Called by the manager when there is a command for this node from the user.
     * Command can be input either from keyboard or file.
     * For now sending messages as ping to a neighbor.
     * @param command The command for this node
     */
    public void onCommand(String command) {
	if(this.matchPingCommand(command)) {
	    return;
	}
	
	logError("Unrecognized command: " + command);
    }

    /**
     * Callback method given to manager to invoke when a timer fires.
     */
    public void pingTimedOut() {
	Iterator iter = this.pings.iterator();
	while(iter.hasNext()) {
	    PingRequest pingRequest = (PingRequest)iter.next();
	    if((pingRequest.getTimeSent() + PingTimeout) < this.manager.now()) {
		try {
		    logOutput("Timing out ping: " + pingRequest);
		    iter.remove();
		}catch(Exception e) {
		    logError("Exception occured while trying to remove an element from ArrayList pings.  Exception: " + e);
		    return;
		}
	    }
	}
	this.addTimer(PingTimeout, "pingTimedOut");
    }

    private boolean matchPingCommand(String command) {
	int index = command.indexOf(" ");
	if(index == -1) {
	    return false;
	}
	try {
	    int destAddr = Integer.parseInt(command.substring(0, index));
	    String message = command.substring(index+1);
	    Packet packet = new Packet(destAddr, this.addr, Packet.MAX_TTL, Protocol.PING_PKT, 0, 
				       Utility.stringToByteArray(message));

	    this.send(destAddr, packet);
	    this.pings.add(new PingRequest(destAddr, Utility.stringToByteArray(message), this.manager.now()));
	    return true;
	}catch(Exception e) {
	    logError("Exception: " + e);
	}
	
	return false;
    }

    private void receivePacket(int from, Packet packet) {
	switch(packet.getProtocol()) {

	case Protocol.PING_PKT:
	    this.receivePing(packet);
	    break;
	    
	case Protocol.PING_REPLY_PKT:
	    this.receivePingReply(packet);
	    break;

	default:
	    logError("Packet with unknown protocol received. Protocol: " + packet.getProtocol());	    
	}
    }

    private void receivePing(Packet packet) {
	logOutput("Received Ping from " + packet.getSrc() + " with message: " + Utility.byteArrayToString(packet.getPayload()));
	
	try {
	    Packet reply = new Packet(packet.getSrc(), this.addr, Packet.MAX_TTL, Protocol.PING_REPLY_PKT, 0, packet.getPayload());
	    this.send(packet.getSrc(), reply);
	}catch(IllegalArgumentException e) {
	    logError("Exception while trying to send a Ping Reply. Exception: " + e);
	}
    }

    // Check that ping reply matches what was sent
    private void receivePingReply(Packet packet) {
	Iterator iter = this.pings.iterator();
	String payload = Utility.byteArrayToString(packet.getPayload());
	while(iter.hasNext()) {
	    PingRequest pingRequest = (PingRequest)iter.next();
	    if( (pingRequest.getDestAddr() == packet.getSrc()) &&
		( Utility.byteArrayToString(pingRequest.getMsg()).equals(payload))) {
		
		logOutput("Got Ping Reply from " + packet.getSrc() + ": " + payload);
		try {
		    iter.remove();
		}catch(Exception e) {
		    logError("Exception occured while trying to remove an element from ArrayList pings while processing Ping Reply.  Exception: " + e);
		}
		return;
	    }
	}
	logError("Unexpected Ping Reply from " + packet.getSrc() + ": " + payload);
    }
    
    private void send(int destAddr, Packet packet) {
	try {
	    this.manager.sendPkt(this.addr, destAddr, packet.pack());
	}catch(IllegalArgumentException e) {
	    logError("Exception: " + e);
	}
    }

    // Adds a timer, to fire in deltaT milliseconds, with a callback to a public function of this class that takes no parameters
    private void addTimer(long deltaT, String methodName) {
	try {
	    Method method = Callback.getMethod(methodName, this, null);
	    Callback cb = new Callback(method, this, null);
	    this.manager.addTimer(this.addr, deltaT, cb);
	}catch(Exception e) {
	    logError("Failed to add timer callback. Method Name: " + methodName +
		     "\nException: " + e);
	}
    }

    private void logError(String output) {
	this.log(output, System.err);
    }

    private void logOutput(String output) {
	this.log(output, System.out);
    }

    private void log(String output, PrintStream stream) {
	stream.println("Node " + this.addr + ": " + output);
    }
}