package testprocs;

import simulator.*;
import modules.color.*;
import modules.consensus.*;
import modules.neighborset.*;
import utils.*;
import java.awt.*;
import java.util.*;

public class ConsensusTest implements ProcessorModule {
  public Symbol getName() { return Symbol.GetSymbol("ConsensusTest"); }
  public void link(PortedProcessor p) {
    pp = p;
    pp.addClockListener(this);
    pp.openNetworkPort(this,uiPort);
    cm = (ColorModule)p.getModule("Color");
    ns = (NeighborSet)p.getModule("NeighborSet");
    cs = (Consensus)p.getModule("Consensus");
  }


  PortedProcessor pp;
  ColorModule cm; NeighborSet ns; Consensus cs;

  static final Symbol uiPort = Symbol.GetSymbol("UI");

  public void init() {
  }

  static int count=0;
  Vector pending = new Vector();

  static int border = 0;
  public boolean signalEvent(Symbol name,Object evtData) {
    boolean dirty = false;
    if(name==uiPort) {
      StringTokenizer tok = new StringTokenizer(evtData.toString());
      if(tok.hasMoreTokens()) { 
        String cmd=tok.nextToken();
        if(cmd.equals("region")) {
          int v = 0;
          if(tok.hasMoreTokens()) v=(new Integer(tok.nextToken())).intValue();
          border = v;
          System.out.println("Consensus borders set to "+v);
        } else if(cmd.equals("join")) {
          Symbol s = Symbol.GetSymbol("X");
          if(tok.hasMoreTokens()) s=Symbol.GetSymbol(tok.nextToken());
          if(!cs.getConsensusProcesses().contains(s)) {
            if(border==0) 
              cs.startConsensus(s,50);
            else 
              cs.startConsensus(s,50,new ConsensusFringe(s,border,pp));
            count++;
          }
        } else if(cmd.equals("leave")) {
          Symbol s = Symbol.GetSymbol("X");
          if(tok.hasMoreTokens()) s=Symbol.GetSymbol(tok.nextToken());
          if(cs.getConsensusProcesses().contains(s)) {
            cs.terminateConsensus(s);
            count--;
          }
        } else if(cmd.equals("propose")) {
          Symbol v = Symbol.GetSymbol("DEF");
          Symbol s = Symbol.GetSymbol("X");
          if(tok.hasMoreTokens()) v=Symbol.GetSymbol(tok.nextToken());
          if(tok.hasMoreTokens()) s=Symbol.GetSymbol(tok.nextToken());
          cs.propose(s,v);
          System.out.println("Process "+s+" attempting consensus on: "+v);
          pending.add(new Tuple(s,v,cs.getConsensus(s).clone()));
        }
      }
    } else if(name==PortedProcessor.kClock) {
      // see if anything pending has been resolved
      Iterator i = pending.iterator();
      while(i.hasNext()) {
        Tuple p = (Tuple)i.next();
        VersionedData oldD = (VersionedData)p.third();
        VersionedData newD = cs.getConsensus(p.first());
        //System.out.println(oldD+"; "+newD);
        if(newD==null || (!newD.equals(oldD) && cs.isConverged(p.first()))) {
          if(newD==null) {
            System.out.println("Monitor detached from process "+p.first()+
                               " attempt to achieve consensus on "+p.second());
          } else { 
            System.out.println("Process "+p.first()+" converged to "+newD);
          }
          i.remove();
        }
      }

      // coloring
      Set s = cs.getConsensusProcesses();
      Object pid = null;
      if(!s.isEmpty()) pid=s.iterator().next();

      if(s.isEmpty()) cm.setColor(Color.blue);
      else if(!cs.containsFull(pid)) cm.setColor(Color.yellow);
      else if(cs.isConverged(pid)) {
        if(false) cm.setColor(Color.green);
        else cm.setColor(cm.numberToColor(cs.getConsensus(pid).getVersion()));
      } else cm.setColor(Color.red);
    }
    return dirty;
  }

  public String toString() { 
    Set s = cs.getConsensusProcesses();
    Object pid = null;
    if(!s.isEmpty()) pid=s.iterator().next();
    String str =  "ConsensusTest: "+count+" members, Consensus: ";
    if(pid!=null) str+=cs.getConsensus(pid);
    else str+="none";
    return str;
  }
}
