gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/LocalDistributionManagerTest.java - geode - Git at Google

 /*=========================================================================
  * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
  * This product is protected by U.S. and international copyright
  * and intellectual property laws. Pivotal products are covered by
  * one or more patents listed at http://www.pivotal.io/patents.
  *=========================================================================
  */
 package com.gemstone.gemfire.distributed.internal;

 //import com.gemstone.gemfire.*;
 //import com.gemstone.gemfire.distributed.DistributedSystem;
 import dunit.*;

 import java.util.*;
 import java.io.*;
 import com.gemstone.gemfire.distributed.internal.membership.*;

 /**
  * This class tests the functionality of a {@link
  * LocalDistributionManager}.
  *
  * @author David Whitlock
  *
  * @since 2.1
  */
 public class LocalDistributionManagerTest
   extends DistributedTestCase {

   public LocalDistributionManagerTest(String name) {
     super(name);
   }

   /**
    * Creates a connection to the distributed system in the given
    * {@link VM}.  Configures the connection for the VM.
    */
   protected void createSystem(VM vm) {
     vm.invoke(new SerializableRunnable("Connect to distributed system") {
         public void run() {
           // @todo davidw Remove reliance on shared memory for
           // configuration
           Properties props = new Properties();
           // props.setProperty(DistributionConfig.ENABLE_SHARED_MEMORY_NAME, "true");
 //           File sysDir =
 //             GemFireConnectionFactory.getInstance().getSystemDirectory();
 //           props.setProperty(DistributionConfig.SYSTEM_DIRECTORY_NAME, sysDir.getPath());
           getSystem(props);
         }
       });
   }

   ////////  Test Methods

   /**
    * Do we see all of the distribution managers in the distributed
    * system?
    */
   public void testCountDMs() {
     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     VM vm1 = host.getVM(1);
     final int systemCount = 2;

     createSystem(vm0);
     vm0.invoke(new SerializableRunnable("Count DMs") {
         public void run() {
           DM dm = getSystem().getDistributionManager();
           assertEquals(systemCount + 1,
                        dm.getNormalDistributionManagerIds().size());
         }
       });
     createSystem(vm1);
     vm1.invoke(new SerializableRunnable("Count DMs Again") {
         public void run() {
           DM dm = getSystem().getDistributionManager();
           assertEquals(systemCount + 2,
                        dm.getNormalDistributionManagerIds().size());
         }
       });
   }

   /**
    * Test that messages that are sent are received in a reasonable
    * amount of time.
    */
   public void testSendMessage() throws InterruptedException {
     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     VM vm1 = host.getVM(1);

     createSystem(vm0);
     createSystem(vm1);

     Thread.sleep(5 * 1000);

     vm0.invoke(new SerializableRunnable("Send message") {
         public void run() {
           DM dm = getSystem().getDistributionManager();
           assertEquals("For DM " + dm.getId(),
                        3, dm.getOtherNormalDistributionManagerIds().size());
           FirstMessage message = new FirstMessage();
           dm.putOutgoing(message);
         }
       });

     Thread.sleep(3 * 1000);

     vm1.invoke(new SerializableRunnable("Was message received?") {
         public void run() {
           WaitCriterion ev = new WaitCriterion() {
             public boolean done() {
               return FirstMessage.received;
             }
             public String description() {
               return null;
             }
           };
           DistributedTestCase.waitForCriterion(ev, 3 * 1000, 200, true);
           FirstMessage.received = false;
         }
       });
   }

   /**
    * Tests the new non-shared {@link ReplyProcessor21}
    */
   public void testReplyProcessor() throws InterruptedException {
     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     VM vm1 = host.getVM(1);

     createSystem(vm0);
     createSystem(vm1);

     Thread.sleep(2 * 1000);

     vm0.invoke(new SerializableRunnable("Send request") {
         public void run() {
           // Send a request, wait for a response
           DM dm = getSystem().getDistributionManager();
           int expected = dm.getOtherNormalDistributionManagerIds().size();
           assertEquals("For DM " + dm.getId(), 3, expected);

           Response.totalResponses = 0;

           Request request = new Request();
           ReplyProcessor21 processor = new ReplyProcessor21(getSystem(),
                dm.getOtherNormalDistributionManagerIds());
           request.processorId = processor.getProcessorId();
           dm.putOutgoing(request);
           try {
             processor.waitForReplies();

           } catch (Exception ex) {
             fail("While waiting for replies", ex);
           }

           assertEquals(expected, Response.totalResponses);
         }
       });

   }

   /** A <code>TestMembershipListener</code> used in this VM */
   protected static TestMembershipListener listener = null;

   /**
    * Does the {@link MembershipListener#memberJoined} method get
    * invoked?
    */
   public void testMemberJoinedAndDeparted()
     throws InterruptedException {

     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     VM vm1 = host.getVM(1);

     createSystem(vm0);
     vm0.invoke(new SerializableRunnable("Install listener") {
         public void run() {
           DM dm = getSystem().getDistributionManager();
           listener = new TestMembershipListener();
           dm.addMembershipListener(listener);
         }
       });
     createSystem(vm1);

     vm0.invoke(new SerializableRunnable("Verify member joining") {
         public void run() {
           WaitCriterion ev = new WaitCriterion() {
             public boolean done() {
               return listener.memberJoined();
             }
             public String description() {
               return null;
             }
           };
           DistributedTestCase.waitForCriterion(ev, 3 * 1000, 200, true);
         }
       });
     vm1.invoke(new SerializableRunnable("Disconnect from system") {
         public void run() {
           getSystem().disconnect();
         }
       });

     vm0.invoke(new SerializableRunnable("Verify member departing") {
         public void run() {
           WaitCriterion ev = new WaitCriterion() {
             public boolean done() {
               return listener.memberDeparted();
             }
             public String description() {
               return null;
             }
           };
           DistributedTestCase.waitForCriterion(ev, 3 * 1000, 200, true);
         }
       });
   }

   /**
    * Tests that the reply processor gets signaled when members go
    * away.
    */
   public void testMembersDepartWhileWaiting()
     throws InterruptedException {

     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     VM vm1 = host.getVM(1);

     createSystem(vm0);
     createSystem(vm1);

     Thread.sleep(3 * 1000);

     AsyncInvocation ai0 =
       vm0.invokeAsync(new SerializableRunnable("Send message and wait") {
           public void run() {
             DM dm = getSystem().getDistributionManager();
             OnlyGFDMReply message = new OnlyGFDMReply();
             ReplyProcessor21 processor = new ReplyProcessor21(getSystem(),
               dm.getOtherNormalDistributionManagerIds());
             message.processorId = processor.getProcessorId();
             dm.putOutgoing(message);

             try {
               processor.waitForReplies();

             } catch (Exception ex) {
               fail("While waiting for replies", ex);
             }
           }
         });

     Thread.sleep(3 * 1000);
     vm1.invoke(new SerializableRunnable("Disconnect from system") {
         public void run() {
           getSystem().disconnect();
         }
       });

     DistributedTestCase.join(ai0, 30 * 1000, getLogWriter());
     if (ai0.exceptionOccurred()) {
       fail("got exception", ai0.getException());
     }
   }

   //////////////////////  Inner Classes  //////////////////////

   /**
    * A message that is send, and when received, sets a
    * <code>boolean</code> static field.
    *
    * @see LocalDistributionManagerTest#testSendMessage
    */
   public static class FirstMessage extends SerialDistributionMessage {

     /** Has a <code>FirstMessage</code> be received? */
     public static boolean received = false;

     public FirstMessage() { }   // For Externalizable

     public void process(DistributionManager dm) {
       received = true;
     }
     public int getDSFID() {
       return NO_FIXED_ID;
     }
   }

   /**
    * A request that is replied to with a {@link Response}
    *
    * @see LocalDistributionManagerTest#testReplyProcessor
    */
   public static class Request extends SerialDistributionMessage
     implements MessageWithReply {

     /** The id of the processor to process the response */
     int processorId;

     public Request() { }        // For Externizable

     public int getProcessorId() {
       return this.processorId;
     }

     /**
      * Reply with a {@link Response}
      */
     public void process(DistributionManager dm) {
       Response response = new Response();
       response.processorId = this.processorId;
       response.setRecipient(this.getSender());
       dm.putOutgoing(response);
     }

     public int getDSFID() {
       return NO_FIXED_ID;
     }

     public void toData(DataOutput out) throws IOException {
       super.toData(out);
       out.writeInt(this.processorId);
     }

     public void fromData(DataInput in)
       throws ClassNotFoundException, IOException {
       super.fromData(in);
       this.processorId = in.readInt();
     }

     public String toString() {
       return "Request with processor " + this.processorId;
     }
   }

   /**
    * A response to a {@link Request}
    *
    * @see LocalDistributionManagerTest#testReplyProcessor
    */
   public static class Response extends SerialDistributionMessage {
     /** The total number of responses that have been received */
     static int totalResponses = 0;

     /** The id of the processor to process the response */
     int processorId;

     public Response() { }       // For Externalizable

     /**
      * Alert the {@link ReplyProcess21} that this reply has been
      * received.
      */
     public void process(DistributionManager dm) {
       // Look up the processor
       ReplyProcessor21 processor =
         ReplyProcessor21.getProcessor(this.processorId);
       assertNotNull("Null processor!", processor);
       synchronized (Response.class) {
         totalResponses++;
       }
       processor.process(this);
     }

     public int getDSFID() {
       return NO_FIXED_ID;
     }

     public void toData(DataOutput out) throws IOException {
       super.toData(out);
       out.writeInt(this.processorId);
     }

     public void fromData(DataInput in)
       throws ClassNotFoundException, IOException {
       super.fromData(in);
       this.processorId = in.readInt();
     }

     public String toString() {
       return "Response with processor " + this.processorId;
     }
   }

   /**
    * A <code>MembershipListener</code> that remembers when members
    * join and depart.
    */
   static class TestMembershipListener implements MembershipListener {

     /** Has a member joined recently? */
     private boolean joined = false;

     /** Has a member departed recently? */
     private boolean departed = false;

     public void memberJoined(InternalDistributedMember id) {
       this.joined = true;
     }

     public void memberDeparted(InternalDistributedMember id, boolean crashed) {
       this.departed = true;
     }

     /**
      * Gets (and then forgets) whether or not a member has recently
      * joined the distributed system.
      */
     public boolean memberJoined() {
       boolean b = this.joined;
       this.joined = false;
       return b;
     }

     public void quorumLost(Set<InternalDistributedMember> failures, List<InternalDistributedMember> remaining) {
     }

     public void memberSuspect(InternalDistributedMember id,
         InternalDistributedMember whoSuspected) {
     }

     /**
      * Gets (and then forgets) whether or not a member has recently
      * departed the distributed system.
      */
     public boolean memberDeparted() {
       boolean b = this.departed;
       this.departed = false;
       return b;
     }
   }

   /**
    * A message that only GemFire distribution managers reply to.
    */
   public static class OnlyGFDMReply extends SerialDistributionMessage
     implements MessageWithReply {

     /** The id of the processor that processes the reply */
     protected int processorId;

     public int getProcessorId() {
       return this.processorId;
     }

     public OnlyGFDMReply() { }  // For Externalizable

     public void process(DistributionManager dm) {
     }

     public int getDSFID() {
       return NO_FIXED_ID;
     }

     public void toData(DataOutput out) throws IOException {
       super.toData(out);
       out.writeInt(this.processorId);
     }

     public void fromData(DataInput in)
       throws ClassNotFoundException, IOException {
       super.fromData(in);
       this.processorId = in.readInt();
     }

     public String toString() {
       return "Only GFDM replies with processor " + this.processorId;
     }
   }

 }
	/*=========================================================================
	* Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
	* This product is protected by U.S. and international copyright
	* and intellectual property laws. Pivotal products are covered by
	* one or more patents listed at http://www.pivotal.io/patents.
	*=========================================================================
	*/
	package com.gemstone.gemfire.distributed.internal;

	//import com.gemstone.gemfire.*;
	//import com.gemstone.gemfire.distributed.DistributedSystem;
	import dunit.*;

	import java.util.*;
	import java.io.*;
	import com.gemstone.gemfire.distributed.internal.membership.*;

	/**
	* This class tests the functionality of a {@link
	* LocalDistributionManager}.
	*
	* @author David Whitlock
	*
	* @since 2.1
	*/
	public class LocalDistributionManagerTest
	extends DistributedTestCase {

	public LocalDistributionManagerTest(String name) {
	super(name);
	}

	/**
	* Creates a connection to the distributed system in the given
	* {@link VM}. Configures the connection for the VM.
	*/
	protected void createSystem(VM vm) {
	vm.invoke(new SerializableRunnable("Connect to distributed system") {
	public void run() {
	// @todo davidw Remove reliance on shared memory for
	// configuration
	Properties props = new Properties();
	// props.setProperty(DistributionConfig.ENABLE_SHARED_MEMORY_NAME, "true");
	// File sysDir =
	// GemFireConnectionFactory.getInstance().getSystemDirectory();
	// props.setProperty(DistributionConfig.SYSTEM_DIRECTORY_NAME, sysDir.getPath());
	getSystem(props);
	}
	});
	}

	//////// Test Methods

	/**
	* Do we see all of the distribution managers in the distributed
	* system?
	*/
	public void testCountDMs() {
	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	VM vm1 = host.getVM(1);
	final int systemCount = 2;

	createSystem(vm0);
	vm0.invoke(new SerializableRunnable("Count DMs") {
	public void run() {
	DM dm = getSystem().getDistributionManager();
	assertEquals(systemCount + 1,
	dm.getNormalDistributionManagerIds().size());
	}
	});
	createSystem(vm1);
	vm1.invoke(new SerializableRunnable("Count DMs Again") {
	public void run() {
	DM dm = getSystem().getDistributionManager();
	assertEquals(systemCount + 2,
	dm.getNormalDistributionManagerIds().size());
	}
	});
	}

	/**
	* Test that messages that are sent are received in a reasonable
	* amount of time.
	*/
	public void testSendMessage() throws InterruptedException {
	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	VM vm1 = host.getVM(1);

	createSystem(vm0);
	createSystem(vm1);

	Thread.sleep(5 * 1000);

	vm0.invoke(new SerializableRunnable("Send message") {
	public void run() {
	DM dm = getSystem().getDistributionManager();
	assertEquals("For DM " + dm.getId(),
	3, dm.getOtherNormalDistributionManagerIds().size());
	FirstMessage message = new FirstMessage();
	dm.putOutgoing(message);
	}
	});

	Thread.sleep(3 * 1000);

	vm1.invoke(new SerializableRunnable("Was message received?") {
	public void run() {
	WaitCriterion ev = new WaitCriterion() {
	public boolean done() {
	return FirstMessage.received;
	}
	public String description() {
	return null;
	}
	};
	DistributedTestCase.waitForCriterion(ev, 3 * 1000, 200, true);
	FirstMessage.received = false;
	}
	});
	}

	/**
	* Tests the new non-shared {@link ReplyProcessor21}
	*/
	public void testReplyProcessor() throws InterruptedException {
	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	VM vm1 = host.getVM(1);

	createSystem(vm0);
	createSystem(vm1);

	Thread.sleep(2 * 1000);

	vm0.invoke(new SerializableRunnable("Send request") {
	public void run() {
	// Send a request, wait for a response
	DM dm = getSystem().getDistributionManager();
	int expected = dm.getOtherNormalDistributionManagerIds().size();
	assertEquals("For DM " + dm.getId(), 3, expected);

	Response.totalResponses = 0;

	Request request = new Request();
	ReplyProcessor21 processor = new ReplyProcessor21(getSystem(),
	dm.getOtherNormalDistributionManagerIds());
	request.processorId = processor.getProcessorId();
	dm.putOutgoing(request);
	try {
	processor.waitForReplies();

	} catch (Exception ex) {
	fail("While waiting for replies", ex);
	}

	assertEquals(expected, Response.totalResponses);
	}
	});

	}

	/** A <code>TestMembershipListener</code> used in this VM */
	protected static TestMembershipListener listener = null;

	/**
	* Does the {@link MembershipListener#memberJoined} method get
	* invoked?
	*/
	public void testMemberJoinedAndDeparted()
	throws InterruptedException {

	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	VM vm1 = host.getVM(1);

	createSystem(vm0);
	vm0.invoke(new SerializableRunnable("Install listener") {
	public void run() {
	DM dm = getSystem().getDistributionManager();
	listener = new TestMembershipListener();
	dm.addMembershipListener(listener);
	}
	});
	createSystem(vm1);

	vm0.invoke(new SerializableRunnable("Verify member joining") {
	public void run() {
	WaitCriterion ev = new WaitCriterion() {
	public boolean done() {
	return listener.memberJoined();
	}
	public String description() {
	return null;
	}
	};
	DistributedTestCase.waitForCriterion(ev, 3 * 1000, 200, true);
	}
	});
	vm1.invoke(new SerializableRunnable("Disconnect from system") {
	public void run() {
	getSystem().disconnect();
	}
	});

	vm0.invoke(new SerializableRunnable("Verify member departing") {
	public void run() {
	WaitCriterion ev = new WaitCriterion() {
	public boolean done() {
	return listener.memberDeparted();
	}
	public String description() {
	return null;
	}
	};
	DistributedTestCase.waitForCriterion(ev, 3 * 1000, 200, true);
	}
	});
	}

	/**
	* Tests that the reply processor gets signaled when members go
	* away.
	*/
	public void testMembersDepartWhileWaiting()
	throws InterruptedException {

	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	VM vm1 = host.getVM(1);

	createSystem(vm0);
	createSystem(vm1);

	Thread.sleep(3 * 1000);

	AsyncInvocation ai0 =
	vm0.invokeAsync(new SerializableRunnable("Send message and wait") {
	public void run() {
	DM dm = getSystem().getDistributionManager();
	OnlyGFDMReply message = new OnlyGFDMReply();
	ReplyProcessor21 processor = new ReplyProcessor21(getSystem(),
	dm.getOtherNormalDistributionManagerIds());
	message.processorId = processor.getProcessorId();
	dm.putOutgoing(message);

	try {
	processor.waitForReplies();

	} catch (Exception ex) {
	fail("While waiting for replies", ex);
	}
	}
	});

	Thread.sleep(3 * 1000);
	vm1.invoke(new SerializableRunnable("Disconnect from system") {
	public void run() {
	getSystem().disconnect();
	}
	});

	DistributedTestCase.join(ai0, 30 * 1000, getLogWriter());
	if (ai0.exceptionOccurred()) {
	fail("got exception", ai0.getException());
	}
	}

	////////////////////// Inner Classes //////////////////////

	/**
	* A message that is send, and when received, sets a
	* <code>boolean</code> static field.
	*
	* @see LocalDistributionManagerTest#testSendMessage
	*/
	public static class FirstMessage extends SerialDistributionMessage {

	/** Has a <code>FirstMessage</code> be received? */
	public static boolean received = false;

	public FirstMessage() { } // For Externalizable

	public void process(DistributionManager dm) {
	received = true;
	}
	public int getDSFID() {
	return NO_FIXED_ID;
	}
	}

	/**
	* A request that is replied to with a {@link Response}
	*
	* @see LocalDistributionManagerTest#testReplyProcessor
	*/
	public static class Request extends SerialDistributionMessage
	implements MessageWithReply {

	/** The id of the processor to process the response */
	int processorId;

	public Request() { } // For Externizable

	public int getProcessorId() {
	return this.processorId;
	}

	/**
	* Reply with a {@link Response}
	*/
	public void process(DistributionManager dm) {
	Response response = new Response();
	response.processorId = this.processorId;
	response.setRecipient(this.getSender());
	dm.putOutgoing(response);
	}

	public int getDSFID() {
	return NO_FIXED_ID;
	}

	public void toData(DataOutput out) throws IOException {
	super.toData(out);
	out.writeInt(this.processorId);
	}

	public void fromData(DataInput in)
	throws ClassNotFoundException, IOException {
	super.fromData(in);
	this.processorId = in.readInt();
	}

	public String toString() {
	return "Request with processor " + this.processorId;
	}
	}

	/**
	* A response to a {@link Request}
	*
	* @see LocalDistributionManagerTest#testReplyProcessor
	*/
	public static class Response extends SerialDistributionMessage {
	/** The total number of responses that have been received */
	static int totalResponses = 0;

	/** The id of the processor to process the response */
	int processorId;

	public Response() { } // For Externalizable

	/**
	* Alert the {@link ReplyProcess21} that this reply has been
	* received.
	*/
	public void process(DistributionManager dm) {
	// Look up the processor
	ReplyProcessor21 processor =
	ReplyProcessor21.getProcessor(this.processorId);
	assertNotNull("Null processor!", processor);
	synchronized (Response.class) {
	totalResponses++;
	}
	processor.process(this);
	}

	public int getDSFID() {
	return NO_FIXED_ID;
	}

	public void toData(DataOutput out) throws IOException {
	super.toData(out);
	out.writeInt(this.processorId);
	}

	public void fromData(DataInput in)
	throws ClassNotFoundException, IOException {
	super.fromData(in);
	this.processorId = in.readInt();
	}

	public String toString() {
	return "Response with processor " + this.processorId;
	}
	}

	/**
	* A <code>MembershipListener</code> that remembers when members
	* join and depart.
	*/
	static class TestMembershipListener implements MembershipListener {

	/** Has a member joined recently? */
	private boolean joined = false;

	/** Has a member departed recently? */
	private boolean departed = false;

	public void memberJoined(InternalDistributedMember id) {
	this.joined = true;
	}

	public void memberDeparted(InternalDistributedMember id, boolean crashed) {
	this.departed = true;
	}

	/**
	* Gets (and then forgets) whether or not a member has recently
	* joined the distributed system.
	*/
	public boolean memberJoined() {
	boolean b = this.joined;
	this.joined = false;
	return b;
	}

	public void quorumLost(Set<InternalDistributedMember> failures, List<InternalDistributedMember> remaining) {
	}

	public void memberSuspect(InternalDistributedMember id,
	InternalDistributedMember whoSuspected) {
	}

	/**
	* Gets (and then forgets) whether or not a member has recently
	* departed the distributed system.
	*/
	public boolean memberDeparted() {
	boolean b = this.departed;
	this.departed = false;
	return b;
	}
	}

	/**
	* A message that only GemFire distribution managers reply to.
	*/
	public static class OnlyGFDMReply extends SerialDistributionMessage
	implements MessageWithReply {

	/** The id of the processor that processes the reply */
	protected int processorId;

	public int getProcessorId() {
	return this.processorId;
	}

	public OnlyGFDMReply() { } // For Externalizable

	public void process(DistributionManager dm) {
	}

	public int getDSFID() {
	return NO_FIXED_ID;
	}

	public void toData(DataOutput out) throws IOException {
	super.toData(out);
	out.writeInt(this.processorId);
	}

	public void fromData(DataInput in)
	throws ClassNotFoundException, IOException {
	super.fromData(in);
	this.processorId = in.readInt();
	}

	public String toString() {
	return "Only GFDM replies with processor " + this.processorId;
	}
	}

	}