import spread.*;
import java.net.InetAddress;
import java.io.ByteArrayOutputStream;

class DbaseGroup implements BasicMessageListener {

  DbaseGroup() throws Exception {
    conn = new SpreadConnection();
    
    int id = -1;

    do {
      String name = "C" + (++id);

      try {
	conn.connect(InetAddress.getByName(null), 0, name, 
				 false, false);
	conn.add(this);
	id = MAX_SPREAD_CLIENTS;
      }

      catch (SpreadException se) { 
	if ( !se.toString().startsWith(SPREAD_DUPLICATE_NAME_ERROR) ) {
	  throw se;
	}
      }
    } while (id < MAX_SPREAD_CLIENTS);
  }

  void sendGet(DbaseHandler handler, byte[] key) throws Exception {
    sendCommand(handler, GET, key, null);
  }

  void sendPut(DbaseHandler handler, byte[] key, byte[] value) throws Exception {
    sendCommand(handler, PUT, key, value);
  }
  
  void sendRemove(DbaseHandler handler, byte[] key) throws Exception {
    sendCommand(handler, REMOVE, key, null);
  }

  // -- BasicMessageListener interface
  
  public void messageReceived(SpreadMessage message) {
    displayMessage(message);

    //
    // Lookup the handler.
    //
    byte[] data = message.getData();
    int tid = get32(data, 0);
    DbaseHandler h = null;

    synchronized (handlers) {
      h = (DbaseHandler) handlers.remove(tid);
    }

    if (h == null) {
      System.out.println("Got unexpected GET_REPLY message");
      return;
    }

    switch (message.getType()) {
    case GET:
      h.getDone(data, 4, data.length - 4);
      break;
      
    case PUT:
      h.putDone();
      break;

    case REMOVE:
      h.removeDone();
      break;
    }
  }

  private void sendCommand(DbaseHandler handler, short op, byte[] key, byte[] value) throws Exception {
    //
    // Create our spread message.
    //
    SpreadMessage msg = new SpreadMessage();
    msg.setSafe();
    msg.addGroup(DBGROUP);
    msg.setType(op);

    // 
    // Set the transaction id
    //
    bout.reset();
    put32(handler.getId(), buf, 0);
    bout.write(buf, 0, 4);

    //
    // Set the key size
    //
    put32(key.length, buf, 0);
    bout.write(buf, 0, 4);

    //
    // Set the key
    //
    bout.write(key);
    
    if (value != null) {
      //
      // Set the value size
      //
      put32(value.length, buf, 0);
      bout.write (buf, 0, 4);
	
      //
      // Set the value
      //
      bout.write (value);
    }

    //
    // And fill our spread message.
    //
    msg.setData(bout.toByteArray());

    //
    // Remember our request handler in order to retrieve it when response is available.
    //
    synchronized (handlers) {
      handlers.put (handler.getId(), handler);
    }
    
    conn.multicast(msg);
  }

  //
  //  Encode a long value in a byte array
  //
  private static int put32 (int value, byte[] array, int offset)
  {
    array[offset++] = (byte) (value) ;
    array[offset++] = (byte) (value >> 8) ;
    array[offset++] = (byte) (value >> 16) ;
    array[offset++] = (byte) (value >> 24) ;
    return offset ;
  }

  //
  //  Retrieve a long value from a byte array
  //
  private static int get32 (byte[] array, int offset)
  {
    int value = ((int) array[offset++]) & 0x0FF ;
    value |= (((int) array[offset++]) << 8) & 0x0FF00 ;
    value |= (((int) array[offset++]) << 16) & 0x0FF0000 ;
    value |= (((int) array[offset++]) << 24) & 0x0FF000000 ;
    return value ;
  }

  private void displayMessage(SpreadMessage msg) {
    try
      {
	if (true)
	  return;

	if(msg.isRegular())
	  {
	    System.out.print("Received a ");
	    if(msg.isUnreliable())
	      System.out.print("UNRELIABLE");
	    else if(msg.isReliable())
	      System.out.print("RELIABLE");
	    else if(msg.isFifo())
	      System.out.print("FIFO");
	    else if(msg.isCausal())
	      System.out.print("CAUSAL");
	    else if(msg.isAgreed())
	      System.out.print("AGREED");
	    else if(msg.isSafe())
	      System.out.print("SAFE");
	    System.out.println(" message.");
				
	    System.out.println("Sent by  " + msg.getSender() + ".");
				
	    System.out.println("Type is " + msg.getType() + ".");
				
	    if(msg.getEndianMismatch() == true)
	      System.out.println("There is an endian mismatch.");
	    else
	      System.out.println("There is no endian mismatch.");
				
	    SpreadGroup groups[] = msg.getGroups();
	    System.out.println("To " + groups.length + " groups.");
				
	    byte data[] = msg.getData();
	    System.out.println("The data is " + data.length + " bytes.");
				
	    System.out.println("The message is: " + new String(data));
	  } else if (msg.isMembership()) {
	    MembershipInfo info = msg.getMembershipInfo();
				
	    if(info.isRegularMembership())
	      {
		SpreadGroup groups[] = msg.getGroups();
				
		System.out.println("Received a REGULAR membership.");
		System.out.println("For group " + info.getGroup() + ".");
		System.out.println("With " + groups.length + " members.");
		System.out.println("I am member " + msg.getType() + ".");
		for(int i = 0 ; i < groups.length ; i++)
		  System.out.println("  " + groups[i]);
				
		System.out.println("Group ID is " + info.getGroupID());
				
		System.out.print("Due to ");
		if(info.isCausedByJoin())
		  {
		    System.out.println("the JOIN of " + info.getJoined() + ".");
		  }
		else if(info.isCausedByLeave())
		  {
		    System.out.println("the LEAVE of " + info.getLeft() + ".");
		  }
		else if(info.isCausedByDisconnect())
		  {
		    System.out.println("the DISCONNECT of " + info.getDisconnected() + ".");
		  }
		else if(info.isCausedByNetwork())
		  {
		    SpreadGroup stayed[] = info.getStayed();
		    System.out.println("NETWORK change.");
		    System.out.println("VS set has " + stayed.length + " members:");
		    for(int i = 0 ; i < stayed.length ; i++)
		      System.out.println("  " + stayed[i]);
		  }
	      }
	    else if(info.isTransition())
	      {
		System.out.println("Received a TRANSITIONAL membership for group " + info.getGroup());
	      }
	    else if(info.isSelfLeave())
	      {
		System.out.println("Received a SELF-LEAVE message for group " + info.getGroup());
	      }
	  } else if ( msg.isReject() ) {
	    // Received a Reject message 
	    System.out.print("Received a ");
	    if(msg.isUnreliable())
	      System.out.print("UNRELIABLE");
	    else if(msg.isReliable())
	      System.out.print("RELIABLE");
	    else if(msg.isFifo())
	      System.out.print("FIFO");
	    else if(msg.isCausal())
	      System.out.print("CAUSAL");
	    else if(msg.isAgreed())
	      System.out.print("AGREED");
	    else if(msg.isSafe())
	      System.out.print("SAFE");
	    System.out.println(" REJECTED message.");
				
	    System.out.println("Sent by  " + msg.getSender() + ".");
				
	    System.out.println("Type is " + msg.getType() + ".");
				
	    if(msg.getEndianMismatch() == true)
	      System.out.println("There is an endian mismatch.");
	    else
	      System.out.println("There is no endian mismatch.");
				
	    SpreadGroup groups[] = msg.getGroups();
	    System.out.println("To " + groups.length + " groups.");
				
	    byte data[] = msg.getData();
	    System.out.println("The data is " + data.length + " bytes.");
				
	    System.out.println("The message is: " + new String(data));
	  } else {
	    System.out.println("Message is of unknown type: " + msg.getServiceType() );
	  }
      }
    catch(Exception e)
      {
	e.printStackTrace();
	System.exit(1);
      }
  }

  //
  // Prefix used to create spread client names. 
  //
  private final static int MAX_SPREAD_CLIENTS = 100000;
    
  //
  // Closed mispelling is mandatory, due to spread. 
  //
  private final static String SPREAD_DUPLICATE_NAME_ERROR = 
    "spread.SpreadException: Connection clsoed during connect attempt to read authnames";

  //
  // Operations sent to the group.
  //
  final static short GET = 1;
  final static short PUT = 2;
  final static short REMOVE = 3;

  //
  // The replicated database group name.
  //
  private final static String DBGROUP = "repldb";
  
  //
  // Spread connection manager.
  //
  SpreadConnection conn;

  //
  // Buffer used to send messages to spread daemon.
  //
  private byte buf[] = new byte [4];
  private ByteArrayOutputStream bout = new ByteArrayOutputStream ();
  
  //
  // List of handlers waiting for their responses.
  //
  private IntHashtable handlers = new IntHashtable();
}
