package org.apache.helix.integration;
   
   /*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you under the Apache License, Version 2.0 (the
    * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   */
  
  import java.util.Date;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.Map.Entry;
  
  import org.apache.helix.TestHelper;
  import org.apache.helix.TestHelper.StartCMResult;
  import org.apache.helix.controller.HelixControllerMain;
  import org.apache.helix.manager.zk.ZNRecordSerializer;
  import org.apache.helix.manager.zk.ZkClient;
  import org.apache.helix.tools.ClusterSetup;
  import org.apache.helix.tools.ClusterStateVerifier;
  import org.apache.helix.tools.ClusterStateVerifier.BestPossAndExtViewZkVerifier;
  import org.apache.helix.tools.ClusterStateVerifier.MasterNbInExtViewVerifier;
  import org.apache.log4j.Logger;
  import org.testng.Assert;
  import org.testng.annotations.AfterClass;
  import org.testng.annotations.BeforeClass;
  
  
  /**
   * 
   * setup a storage cluster and start a zk-based cluster controller in stand-alone mode
   * start 5 dummy participants verify the current states at end
   */
  
  public class ZkStandAloneCMTestBase extends ZkIntegrationTestBase
  {
    private static Logger                LOG               =
                                                               Logger.getLogger(ZkStandAloneCMTestBase.class);
  
    protected static final int           NODE_NR           = 5;
    protected static final int           START_PORT        = 12918;
    protected static final String        STATE_MODEL       = "MasterSlave";
    protected static final String        TEST_DB           = "TestDB";
    protected static final int           _PARTITIONS       = 20;
  
    protected ClusterSetup               _setupTool        = null;
    protected final String               CLASS_NAME        = getShortClassName();
    protected final String               CLUSTER_NAME      = CLUSTER_PREFIX + "_"
                                                               + CLASS_NAME;
  
    protected Map<String, StartCMResult> _startCMResultMap =
                                                               new HashMap<String, StartCMResult>();
    protected ZkClient                   _zkClient;
    
    int _replica = 3;
  
    @BeforeClass
    public void beforeClass() throws Exception
    {
  //    Logger.getRootLogger().setLevel(Level.INFO);
      System.out.println("START " + CLASS_NAME + " at "
          + new Date(System.currentTimeMillis()));
  
      _zkClient = new ZkClient(ZK_ADDR);
      _zkClient.setZkSerializer(new ZNRecordSerializer());
      String namespace = "/" + CLUSTER_NAME;
      if (_zkClient.exists(namespace))
      {
        _zkClient.deleteRecursive(namespace);
      }
      _setupTool = new ClusterSetup(ZK_ADDR);
  
      // setup storage cluster
      _setupTool.addCluster(CLUSTER_NAME, true);
      _setupTool.addResourceToCluster(CLUSTER_NAME, TEST_DB, _PARTITIONS, STATE_MODEL);
      for (int i = 0; i < NODE_NR; i++)
      {
        String storageNodeName = PARTICIPANT_PREFIX + "_" + (START_PORT + i);
        _setupTool.addInstanceToCluster(CLUSTER_NAME, storageNodeName);
      }
      _setupTool.rebalanceStorageCluster(CLUSTER_NAME, TEST_DB, _replica);
  
      // start dummy participants
      for (int i = 0; i < NODE_NR; i++)
      {
       String instanceName = PARTICIPANT_PREFIX + "_" + (START_PORT + i);
       if (_startCMResultMap.get(instanceName) != null)
       {
         LOG.error("fail to start particpant:" + instanceName
             + "(participant with same name already exists)");
       }
       else
       {
         StartCMResult result =
             TestHelper.startDummyProcess(ZK_ADDR, CLUSTER_NAME, instanceName);
         _startCMResultMap.put(instanceName, result);
       }
     }
 
     // start controller
     String controllerName = CONTROLLER_PREFIX + "_0";
     StartCMResult startResult =
         TestHelper.startController(CLUSTER_NAME,
                                    controllerName,
                                    ZK_ADDR,
                                    HelixControllerMain.STANDALONE);
     _startCMResultMap.put(controllerName, startResult);
     
     boolean result =
         ClusterStateVerifier.verifyByZkCallback(new MasterNbInExtViewVerifier(ZK_ADDR,
                                                                               CLUSTER_NAME));
 
     result =
         ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
                                                                                  CLUSTER_NAME));
     Assert.assertTrue(result);
   }
 
   @AfterClass
   public void afterClass() throws Exception
   {
     /**
      * shutdown order: 1) disconnect the controller 2) disconnect participants
      */
    
     StartCMResult result;
     Iterator<Entry<String, StartCMResult>> it = _startCMResultMap.entrySet().iterator();
     while (it.hasNext())
     {
       String instanceName = it.next().getKey();
       if (instanceName.startsWith(CONTROLLER_PREFIX))
       {
         result = _startCMResultMap.get(instanceName);
         result._manager.disconnect();
         result._thread.interrupt();
         it.remove();
       }
     }
 
     Thread.sleep(100);
     it = _startCMResultMap.entrySet().iterator();
     while (it.hasNext())
     {
       String instanceName = it.next().getKey();
       result = _startCMResultMap.get(instanceName);
       result._manager.disconnect();
       result._thread.interrupt();
       it.remove();
     }
 
     _zkClient.close();
     // logger.info("END at " + new Date(System.currentTimeMillis()));
     System.out.println("END " + CLASS_NAME + " at "
         + new Date(System.currentTimeMillis()));
   }
 }