[jgroups-dev] Advice on best practices for implementing a new protocol

package org.jgroups.protocols.jentMESH;

import org.jgroups.*;
import org.jgroups.annotations.*;
import org.jgroups.protocols.RELAY.RelayHeader;
import org.jgroups.stack.Protocol;
import org.jgroups.util.TimeScheduler;
import org.jgroups.util.Util;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Protocol to establish groups where all nodes are not connected to all other nodes.
 * See design document located at: http://archive.jentfoo.com/coding_projects/jentMesh_design.pdf
 *
 * @author Mike Jensen
 */
@Experimental @Unsupported
@MBean(description="TREEMESH protocol")
public class TREEMESH extends Protocol {
  /* ------------------------------------------    Properties     ---------------------------------------------- */
  @Property(description="How frequently in milliseconds should the node broadcast it's heartbeat")
  protected int broadcastIntravel = 1000;

  @Property(description="How long till we timeout a silent node, should be above the heartbeat")
  protected int nodeTimeout = broadcastIntravel * 2;

  @Property(description="How many connections should each peer try to maintain")
  protected short peerConMax = 3;

  @Property(description="Number representing how sure we want to be before we make decisions")
  protected final short sampleSize = 5;
  
  @Property(description="Time delay between regular process calls in milliseconds")
  protected final long processPeriod = 500;
  
  /* ---------------------------------------------    Fields    ------------------------------------------------ */
  protected Address local_addr;

  @ManagedAttribute
  protected volatile boolean is_coord = false;  // TODO - is_coord = Util.isCoordinator(view, local_addr);

  protected TimeScheduler timer;
  private MeshModel meshModel;
  
  public void init() throws Exception {
    meshModel = new MeshModel(sampleSize);
    timer = getTransport().getTimer();
    
    // schedule a thread to do regular tasks
    timer.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        maybeMakePeerConnections();
        
        maybeMakeLeafConnections();
        
        removeDuplicateConnections();

        timeoutDeadNodes();
      }
    }, 500, processPeriod, TimeUnit.MILLISECONDS);
    
    // schedule a thread to send the heartbeat
    timer.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        // TODO - send heartbeat
      }
    }, 500, broadcastIntravel, TimeUnit.MILLISECONDS);
  }

  protected void maybeMakePeerConnections() {
    /* TODO - if no peer connections exist, 
     *        connect to a peer recommended from the MeshModel */
  }

  protected void maybeMakeLeafConnections() {
    /* TODO - if one peer connection exists, and we have less than peerConMax - 1 leaf connections, 
     *        establish a connection to the most distant leaf */
  }

  protected void removeDuplicateConnections() {
    /* TODO - cycle through all connections, 
     *        if there is a duplicate connection and the hash of our address is lower than the remote address, 
     *        remove the connection 
     *        (if it is higher, we expect the remote peer will destroy the connection, 
     *        this way both connections are not destroyed at the same time) */
  }
  
  protected void timeoutDeadNodes() {
    /* TODO - Cycle through all known nodes, 
     *        if we have not heard any heartbeats or messages that are from or passed through that node, 
     *        assume they are dead and take appropriate action */
  }

  public void stop() {
  }
  
  public Object down(Event event) {
    switch (event.getType()) {
      case Event.MSG:
        // add our header on the message before passing on
        Message msg = (Message)event.getArg();
        msg.putHeaderIfAbsent(getId(), new TreeMeshHeader(local_addr, 
                                                          meshModel.getAvgHops(), 
                                                          meshModel.peerConnectionCount(), 
                                                          meshModel.leafConnectionCount()));
        break;
      case Event.DISCONNECT:
        break;
      case Event.SET_LOCAL_ADDRESS:
        local_addr = (Address) event.getArg();
        break;
      default: 
        log.warn("Got unhandled down event: " + event.toString());  // TODO - remove temp logging
        break;
    }
    return down_prot.down(event);
  }
  
  public Object up(Event event) {
    switch (event.getType()) {
      case Event.MSG:
          Message msg = (Message)event.getArg();
          Address dest = msg.getDest();
          TreeMeshHeader hdr = (TreeMeshHeader)msg.getHeader(getId());
          
          // analyze the message for more informatio
          meshModel.analyzeMessageHopRecord(hdr.hopRecord);
          
          // TODO - forward the msg to other connected nodes as needed
          // TODO - forward to application if needed
        break;
      default: 
        log.warn("Got unhandled up event: " + event.toString());  // TODO - remove temp logging
        break;
    }
    return up_prot.up(event);
  }
  
  public static class TreeMeshHeader extends Header {
    private List<MsgHop> hopRecord;
    
    private TreeMeshHeader() {
      hopRecord = new LinkedList<MsgHop>();
    }
    
    public TreeMeshHeader(Address local_addr, short avgHops, short peerConnections, short leafConnections) {
      this();
      addHop(local_addr, avgHops, peerConnections, leafConnections);
    }
    
    public int size() {
      // hop entries * size of entry
      return hopRecord.size() * (((Short.SIZE / 8) * 3) + 1); // TODO - this is not very accurate
    }
    
    public void writeTo(DataOutputStream out) throws IOException {
    }
    
    public void readFrom(DataInputStream in) throws IOException, IllegalAccessException,
        InstantiationException {
    }
    
    public void addHop(Address local_addr, short avgHops, short peerConnections, short leafConnections) {
      if (local_addr != null) {
        hopRecord.add(new MsgHop(local_addr, avgHops, peerConnections, leafConnections));
      } else {
        // TODO - log something scary
      }
    }
    
    public class MsgHop implements Serializable {
      public final Address host;
      public final short avgHops;
      public final short peerConnections;
      public final short leafConnections;
      
      public MsgHop(Address host, short avgHops, short peerConnections, short leafConnections) {
        this.host = host;
        this.avgHops = avgHops;
        this.peerConnections = peerConnections;
        this.leafConnections = leafConnections;
      }
    }
  }
}