Follow terracotta-api#117 changes

Author: ramsai1729

.gitignore

@@ -1,5 +1,7 @@
 target/
 
+*.iml
 .project
 .classpath
+.idea/
 .settings/

galvan-support/src/main/java/org/terracotta/testing/client/IPCClusterControl.java

@@ -62,6 +62,16 @@ public void waitForActive() throws Exception {
     ipcManager.waitForActive();
   }
 
+  @Override
+  public void terminateActive() throws Exception {
+    ipcManager.terminateActive();
+  }
+
+  @Override
+  public void startLastTerminatedServer() throws Exception {
+    ipcManager.startLastTerminatedServer();
+  }
+
   @Override
   public void waitForPassive() throws Exception {
     ipcManager.waitForPassive();

galvan-support/src/main/java/org/terracotta/testing/rules/BasicExternalCluster.java

@@ -197,6 +197,16 @@ public void waitForActive() throws Exception {
         cluster.stripeControl.waitForActive();
       }
 
+      @Override
+      public void terminateActive() throws Exception {
+        cluster.stripeControl.terminateActive();
+      }
+
+      @Override
+      public void startLastTerminatedServer() throws Exception {
+        cluster.stripeControl.startLastTerminatedServer();
+      }
+
       @Override
       public void waitForPassive() throws Exception {
         cluster.stripeControl.waitForPassive();

galvan/src/main/java/org/terracotta/testing/client/ClientSideIPCManager.java

@@ -84,6 +84,10 @@ public void restartActive() {
     sendAndWait(IPCMessageConstants.RESTART_ACTIVE);
   }
 
+  public void startLastTerminatedServer() { sendAndWait(IPCMessageConstants.START_LAST_TERMINATED_ACTIVE); }
+
+  public void terminateActive() { sendAndWait(IPCMessageConstants.TERMINATE_ACTIVE); }
+
   public void shutDownStripeAndWaitForTermination() {
     sendAndWait(IPCMessageConstants.SHUT_DOWN_STRIPE);
   }

galvan/src/main/java/org/terracotta/testing/common/IPCMessageConstants.java

@@ -35,6 +35,14 @@ public static String ackFrom(String message) {
   public static final String RESTART_ACTIVE_SYN = synFrom(RESTART_ACTIVE);
   public static final String RESTART_ACTIVE_ACK = ackFrom(RESTART_ACTIVE);
 
+  public static final String TERMINATE_ACTIVE = "TERMINATE_ACTIVE";
+  public static final String TERMINATE_ACTIVE_SYN = synFrom(TERMINATE_ACTIVE);
+  public static final String TERMINATE_ACTIVE_ACK = ackFrom(TERMINATE_ACTIVE);
+
+  public static final String START_LAST_TERMINATED_ACTIVE = "START_LAST_TERMINATE_ACTIVE";
+  public static final String START_LAST_TERMINATED_ACTIVE_SYN = synFrom(START_LAST_TERMINATED_ACTIVE);
+  public static final String START_LAST_TERMINATED_ACTIVE_ACK = ackFrom(START_LAST_TERMINATED_ACTIVE);
+
   public static final String SHUT_DOWN_STRIPE = "SHUT_DOWN_STRIPE";
   public static final String SHUT_DOWN_STRIPE_SYN = synFrom(SHUT_DOWN_STRIPE);
   public static final String SHUT_DOWN_STRIPE_ACK = ackFrom(SHUT_DOWN_STRIPE);

galvan/src/main/java/org/terracotta/testing/master/ClientEventManager.java

@@ -58,6 +58,28 @@ public void onEvent(Event e) throws Throwable {
         processStdin.println(IPCMessageConstants.RESTART_ACTIVE_ACK);
         processStdin.flush();
       }});
+
+    String terminateActiveEventName = "Terminate active";
+    eventMap.put(IPCMessageConstants.TERMINATE_ACTIVE_SYN, terminateActiveEventName);
+    subBus.on(terminateActiveEventName, new EventListener() {
+      @Override
+      public void onEvent(Event e) throws Throwable {
+        control.terminateActive();
+        // We also want to send the ACK to the client.
+        processStdin.println(IPCMessageConstants.TERMINATE_ACTIVE_ACK);
+        processStdin.flush();
+      }});
+
+    String startLastTerminatedServerEventName = "Start Last Terminated active";
+    eventMap.put(IPCMessageConstants.START_LAST_TERMINATED_ACTIVE_SYN, startLastTerminatedServerEventName);
+    subBus.on(startLastTerminatedServerEventName, new EventListener() {
+      @Override
+      public void onEvent(Event e) throws Throwable {
+        control.startLastTerminatedServer();
+        // We also want to send the ACK to the client.
+        processStdin.println(IPCMessageConstants.START_LAST_TERMINATED_ACTIVE_ACK);
+        processStdin.flush();
+      }});
 
     String shutDownStripeEventName = "Shut down stripe";
     eventMap.put(IPCMessageConstants.SHUT_DOWN_STRIPE_SYN, shutDownStripeEventName);

galvan/src/main/java/org/terracotta/testing/master/IMultiProcessControl.java

@@ -28,6 +28,10 @@ public interface IMultiProcessControl {
 
   public void restartActive();
 
+  public void terminateActive();
+
+  public void startLastTerminatedServer();
+
   public void shutDown();
 
   public void waitForActive();

galvan/src/main/java/org/terracotta/testing/master/MultiStripeProcessControl.java

@@ -41,6 +41,20 @@ public synchronized void restartActive() {
     }
   }
 
+  @Override
+  public void terminateActive() {
+    for (IMultiProcessControl oneControl : this.subControl) {
+      oneControl.terminateActive();
+    }
+  }
+
+  @Override
+  public void startLastTerminatedServer() {
+    for (IMultiProcessControl oneControl : this.subControl) {
+      oneControl.startLastTerminatedServer();
+    }
+  }
+
   @Override
   public synchronized void shutDown() {
     for (IMultiProcessControl oneControl : this.subControl) {

galvan/src/main/java/org/terracotta/testing/master/SynchronousProcessControl.java

@@ -16,7 +16,9 @@
 package org.terracotta.testing.master;
 
 import java.io.IOException;
+import java.util.EmptyStackException;
 import java.util.List;
+import java.util.Stack;
 import java.util.Vector;
 
 import org.terracotta.testing.common.Assert;
@@ -34,6 +36,7 @@ public class SynchronousProcessControl implements IMultiProcessControl {
   private final List<ServerProcess> unknownServers = new Vector<ServerProcess>();
   // It is invalid to use a process control object which has already been shut down so keep that flag.
   private boolean isShutDown;
+  private Stack<ServerProcess> terminatedServers = new Stack();
 
   public SynchronousProcessControl(ITestStateManager stateManager, ContextualLogger logger) {
     this.stateManager = stateManager;
@@ -84,6 +87,60 @@ public synchronized void restartActive() {
     this.logger.output("<<< restartActive");
   }
 
+  @Override
+  public void terminateActive() {
+    this.logger.output(">>> terminateActive");
+    verifyNotShutdown();
+    // First, make sure that there is an active.
+    internalWaitForActive();
+    // We MUST now have an active.
+    Assert.assertTrue(null != this.activeServer);
+
+    // Remove the active.
+    ServerProcess victim = this.activeServer;
+    this.activeServer = null;
+    // Stop it.
+    try {
+      int ret = victim.stop();
+      this.logger.output("terminated server, return status: " + ret);
+    } catch (InterruptedException e) {
+      // We can't leave a consistent state if interrupted at this point.
+      Assert.unexpected(e);
+    }
+    // Return the process to its installation and get a new one.
+    ServerInstallation underlyingInstallation = victim.getUnderlyingInstallation();
+    underlyingInstallation.retireProcess(victim);
+
+    // Close its logs.
+    try {
+      underlyingInstallation.closeStandardLogFiles();
+    } catch (IOException e) {
+      // We don't expect this IOException on closing the logs.
+      Assert.unexpected(e);
+    }
+  }
+
+  @Override
+  public void startLastTerminatedServer() {
+    try {
+      ServerProcess lastTerminatedServer = terminatedServers.pop();
+
+      ServerInstallation underlyingInstallation = lastTerminatedServer.getUnderlyingInstallation();
+      ServerProcess freshProcess = underlyingInstallation.createNewProcess(this.stateManager);
+
+      // Start it.
+      long pid = freshProcess.start();
+      this.logger.output("Server restarted with PID: " + pid);
+      // Enqueue it onto the unknown list.
+      this.unknownServers.add(freshProcess);
+
+      // At this point, we don't know the active server.
+      this.logger.output("<<< restartActive");
+    } catch (EmptyStackException e) {
+      throw new RuntimeException("There are no terminated processes");
+    }
+  }
+
   @Override
   public synchronized void shutDown() {
     this.logger.output(">>> shutDown");
@@ -175,14 +232,23 @@ private void internalWaitForActive() {
       waitForAllUnknowns();
       // If we still have no active, walk the passives to see if one was promoted.
       if (null == this.activeServer) {
-        // Note that we don't want to walk this list while placing servers in states since looping the list, at this point, is either a bug or a serious race condition.
-        Vector<ServerProcess> oldPassives = new Vector<ServerProcess>(this.passiveServers);
-        this.passiveServers.clear();
-        while (!oldPassives.isEmpty()) {
-          ServerProcess passive = oldPassives.remove(0);
-          waitAndPlaceServerInState(passive);
+        checkAllPassivesState();
+
+        // If we still can't find an active, do multiple retries with some timeout as it might take sometime for
+        // other passives to become active (Note that this is a hacky way to fix the issue for now, we are going to
+        // address this properly later)
+        if(null == this.activeServer) {
+          int retryCount = 30;
+          while (retryCount-- != 0) {
+            try {
+              Thread.sleep(1000);
+              checkAllPassivesState();
+            } catch (InterruptedException e) {
+              Assert.unexpected(e);
+            }
+          }
         }
-        
+
         // See if we are still without an active.
         if (null == this.activeServer) {
           // If there is no active at this point, it probably means that there is something seriously wrong with the
@@ -193,6 +259,16 @@ private void internalWaitForActive() {
     }
   }
 
+  private void checkAllPassivesState() {
+    // Note that we don't want to walk this list while placing servers in states since looping the list, at this point, is either a bug or a serious race condition.
+    Vector<ServerProcess> oldPassives = new Vector<ServerProcess>(this.passiveServers);
+    this.passiveServers.clear();
+    while (!oldPassives.isEmpty()) {
+      ServerProcess passive = oldPassives.remove(0);
+      waitAndPlaceServerInState(passive);
+    }
+  }
+
   private void waitForAllUnknowns() {
     while (!this.unknownServers.isEmpty()) {
       ServerProcess unknown = this.unknownServers.remove(0);