/**
 * Ivy java Async API tester: bindAsyncMsg, sendAsyncMsg.
 *
 * @author  Yannick Jestin <mailto:jestin@cena.fr>
 *
 * (c) CENA
 *
 * usage: java AsyncAPI -h
 *
 * this program tests the Asynchronous reception of messages ( each callback
 * is performed in a separate threads ). It also exhibits the behaviour of the
 * library with regards to concurrent connections ! To stress the test, try it
 * with different JVM ( kaffe is especially hard to pass ), and on SMP
 * machines ...
 *
 * changelog
 *
 *  1.2.6 : async sending seems utterly buggy ...
 *
 */
import fr.dgac.ivy.*;
import gnu.getopt.*;

class AsyncAPI {

  public static final int MSGSIZE = 10;
  public static final int NBITER = 100;
  public static final int DELAYMS = 1000;
  public static final String HEADER = "ASYNCPACKET";
  public static final String TOSUBSCRIBE = "^"+HEADER+"([0-9]*) (.*)";
  public static final String RECEIVENAME = "MSreceive";
  public static final String SENDNAME = "MSsend";
  private static long epoch = System.currentTimeMillis();

  private Ivy bus;
  DelayAnswer delay;
  int re;
  private String name;
  boolean verbose;
  private int nbpacket;
  private int count=0,total=0;
  protected Integer truc = new Integer(0);

  public AsyncAPI(int nb,String domain,int d, boolean v,boolean async) throws IvyException {
    verbose=v;
    nbpacket=nb;
    name = "MSreceive";
    bus = new Ivy(name,name+" Ready", null);
    delay=new DelayAnswer(d);
    if (async)
      re = bus.bindAsyncMsg(TOSUBSCRIBE,delay);
    else
      re = bus.bindMsg(TOSUBSCRIBE,delay);
    bus.start(domain);
  }

  private static java.text.DateFormat df = java.text.DateFormat.getTimeInstance();
  private static String date() {
    return  "["+df.format(new java.util.Date())+"] ";
  }

  class DelayAnswer implements IvyMessageListener {
    int delay;
    public DelayAnswer(int delay) {
      this.delay=delay;
    }
    public void receive(IvyClient ic, String[] args) {
      synchronized(truc) {
	count++;
	if (verbose) { 
	  System.out.println(date()+count+"/"+nbpacket+" packets received ("+args[0]+")");
	  total+=Integer.parseInt(args[0]);
	}
	if (count<nbpacket) return;
	if (total==(((nbpacket+1)*nbpacket)/2)) {
	  System.out.println("receiver quitting the bus normally");
	  ic.sendDie("ok, bye");
	  bus.stop();
	} else {
	  System.out.println("wrong count and total, hit ^C to exit");
	  //System.exit(-1);
	}
      }
      try { Thread.sleep(delay); } catch (InterruptedException ie) { }
    }
  }

  public static final String helpmsg = "usage: java "+SENDNAME+" [options]\n\t-b domain\n\t-r\t to enable async reception\n\t-x to enable async sending\n\t-l loop count (default "+NBITER+")\n\t-s msgsize\t int value (default "+MSGSIZE+")\n\t-d delay in milliseconds (default "+DELAYMS+")\n\t-q \tquiet\n\t-e\tsend bus stop at the end\n\t-h\thelp\n\n";
  public static void main(String[] args) throws IvyException {
    Getopt opt = new Getopt(SENDNAME,args,"b:l:d:s:xrqeh");
    String domain=Ivy.getDomain(null); // default bus
    int nb = NBITER, delay = DELAYMS, c, size = MSGSIZE;
    boolean doasyncSend=false, doasyncBind=false, verbose=true, exit=false;
    while ((c = opt.getopt()) != -1) switch (c) {
      case 'b':
	domain=opt.getOptarg();
	break;
      case 's':
	size=Integer.parseInt(opt.getOptarg());
	break;
      case 'l':
	nb=Integer.parseInt(opt.getOptarg());
	break;
      case 'd':
	delay=Integer.parseInt(opt.getOptarg());
	break;
      case 'e':
	exit=true;
	break;
      case 'q':
	verbose=false;
	break;
      case 'x':
	System.out.println("async sending is not robust enough. end of test.");
	System.exit(-1);
	doasyncSend=true;
	break;
      case 'r':
	doasyncBind=true;
	break;
    case 'h':
    default:
	System.out.println(helpmsg);
	System.exit(0);
    }
    System.out.println("bus:"+domain+" loop:"+nb+" delay:"+delay+" verbose:"+verbose+" asyncBind:"+
	doasyncBind+" asyncSend:"+doasyncSend+" msgsize:"+size);

    AsyncAPI receiver = new AsyncAPI(nb,domain,delay,verbose,doasyncBind);
    Ivy mainbus = new Ivy(SENDNAME,SENDNAME+" Ready", null);
    mainbus.start(domain);
    if ((mainbus.waitForClient(RECEIVENAME,5000))==null) {
      System.out.println(RECEIVENAME+" did not join the bus. Quitting");
      System.exit(0);
    }
    System.out.println(RECEIVENAME+" is here, sending packets");
    StringBuffer tosend = new StringBuffer(size);
    for (int i=0;i<size;i++) tosend.append("a");
    for (int i=1;i<=nb;i++) {
      if (verbose) System.out.println(date()+"sending packet "+i);
      mainbus.sendMsg(HEADER+i+" "+tosend.toString());
      // mainbus.sendMsg(HEADER+i+" "+tosend.toString(),doasyncSend);
    }
    System.out.println(date()+"sender has sent all its packets, waiting for a die message");
    // i won't stop the sender's bus here, otherwise the all the packet
    // can still be unprocessed
    // TODO regession test for Ivy.stop()
  }

}