/**
 * This function puts portfolio objects into the created Region (PR or Local)
 * *
 */
public CacheSerializableRunnable getCacheSerializableRunnableForPRPuts(final String regionName,
    final Object[] portfolio, final int from, final int to) {
  SerializableRunnable puts = new CacheSerializableRunnable("Region Puts") {
    @Override
    public void run2() throws CacheException {
      Cache cache = CacheFactory.getAnyInstance();
      Region region = cache.getRegion(regionName);
      for (int j = from; j < to; j++)
        region.put(new Integer(j), portfolio[j]);
      getLogWriter()
          .info(
              "PRQueryDUnitHelper#getCacheSerializableRunnableForPRPuts: Inserted Portfolio data on Region "
                  + regionName);
    }
  };
  return (CacheSerializableRunnable) puts;
}
 

public void waitForValue(VM vm, final String key, final int value) {
  SerializableCallable createSystem = new SerializableCallable() {
    public Object call() throws Exception {
      Cache cache = getCache();
      final Region region1 = cache.getRegion("region");
      waitForCriterion(new WaitCriterion() {

        public String description() {
          return "Didn't receive update over the WAN";
        }

        public boolean done() {
          return region1.get(key) != null;
        }
        
      }, 30000, 100, true);
      assertEquals(new SimpleClass(value, (byte) value), region1.get(key));
      return null;
    }
    
  };
  vm.invoke(createSystem);
}
 

public void testLikeWithLimitWithParameter() throws Exception {
  String queryString = "SELECT DISTINCT entry FROM $1 entry WHERE entry.key like $2 ORDER BY entry.key LIMIT $3 ";
  SelectResults result;
  Region region = CacheUtils.createRegion("portfolios1", Portfolio.class);
  for (int i = 0; i < 100; i++) {
    region.put( "p"+i, new Portfolio(i));
  }
  
  Object[] params = new Object[3];
  params[0] = region.entrySet();
  params[1] = "p%";
  params[2] = 5;

  SelectResults results = (SelectResults)qs.newQuery(queryString).execute(params);
  assertEquals(5, results.size());
}
 

protected static void validateGetStats(int expectedVal) {

    GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
    Region r = cache.getRegion("/TRADE/CUSTOMERS");
    assertNotNull(r);
    assert (r instanceof PartitionedRegion);

    PartitionedRegion parRegion = (PartitionedRegion) r;
    int val = parRegion.getPrStats().getStats().get("getsCompleted").intValue();
    assertEquals(expectedVal, val);
    
    Region r1 = cache.getRegion("/TRADE/PROSPECTIVES");
    
    assert (r1 instanceof DistributedRegion);
    
    LocalRegion lRegion = (LocalRegion)r1;
    val = lRegion.getCachePerfStats().getStats().get("gets").intValue();
    assertEquals(expectedVal, val);

  }
 

protected static String getDiskAttributes(Region r) {
  String dsn = r.getAttributes().getDiskStoreName();
  if (dsn == null) {
    throw new HydraConfigException("Persistence is not configured");
  }
  DiskStore ds = DiskStoreHelper.getDiskStore(dsn);
  return "the DiskStore is" + ds
       + ", the allow force compaction is "
       + ds.getAllowForceCompaction()
       + ", the auto compact is "
       + ds.getAutoCompact()
       + ", the compaction threshold is "
       + ds.getCompactionThreshold()
       + ", the disk dir num is "
       + ds.getDiskDirs().length
       + ", the max oplog size is "
       + ds.getMaxOplogSize()
       + ", the queue size is "
       + ds.getQueueSize()
       + ", the synchronous (from the region) is "
       + r.getAttributes().isDiskSynchronous()
       ;
}
 

/**
 *
 * @param member
 *          member to which this MBean belongs
 * @param monitoringRegion
 *          corresponding MonitoringRegion
 * @param objectName
 *          ObjectName of the MBean
 * @param interfaceClass
 *          on which interface the proxy to be exposed
 * @return Object
 * @throws ClassNotFoundException
 * @throws IntrospectionException
 */
public static Object newProxyInstance(DistributedMember member,
    Region<String, Object> monitoringRegion, ObjectName objectName,
    Class interfaceClass) throws ClassNotFoundException,
    IntrospectionException {
  boolean isMXBean = JMX.isMXBeanInterface(interfaceClass);
  boolean notificationBroadcaster = ((FederationComponent) monitoringRegion
      .get(objectName.toString())).isNotificationEmitter();

  InvocationHandler handler = new MBeanProxyInvocationHandler(member,
      objectName, monitoringRegion, isMXBean);

  Class[] interfaces;

  if (notificationBroadcaster) {
    interfaces = new Class[] { interfaceClass, ProxyInterface.class,
        NotificationBroadCasterProxy.class };
  } else {
    interfaces = new Class[] { interfaceClass, ProxyInterface.class };
  }

  Object proxy = Proxy.newProxyInstance(MBeanProxyInvocationHandler.class
      .getClassLoader(), interfaces, handler);

  return interfaceClass.cast(proxy);
}
 

public static void registerInterest()
{
  cache.getLogger().info(
      "<ExpectedException action=add>" + "RegionDestroyedException"
      + "</ExpectedException>");
  try {
    Region r = cache.getRegion("/" + regionName);
    assertNotNull(r);
    List listOfKeys = new ArrayList();
    listOfKeys.add("key-1");
    listOfKeys.add("key-2");
    listOfKeys.add("key-3");
    listOfKeys.add("key-4");
    listOfKeys.add("key-5");
    r.registerInterest(listOfKeys);
    fail("expected RegionDestroyedException");
  } catch (ServerOperationException expected) {
    assertEquals(RegionDestroyedException.class, expected.getCause().getClass());
  }
  finally {
    cache.getLogger().info(
        "<ExpectedException action=remove>" + "RegionDestroyedException"
        + "</ExpectedException>");
  }
}
 

/** 
 *  Hydra task to create a BridgeServer, creates the cache & regions to act as
 *  parents of dynamically created regions.
 */
public synchronized static void HydraTask_initBridgeServer() {

   // configure DynamicRegionFactory - required before cache is open
   File d = new File("DynamicRegionData" + ProcessMgr.getProcessId());
   d.mkdirs();
   DynamicRegionFactory.get().open(new DynamicRegionFactory.Config(d, null));

   // create cache
   CacheHelper.createCache("bridge");

   // create regions to act as parent(s) of dynamic regions
   int numRoots = TestConfig.tab().intAt(DynamicRegionPrms.numRootRegions);
   int breadth = TestConfig.tab().intAt(DynamicRegionPrms.numSubRegions);
   int depth = TestConfig.tab().intAt(DynamicRegionPrms.regionDepth);

   for (int i=0; i<numRoots; i++) {
     String rootName = "root" + (i+1);
     Region rootRegion = RegionHelper.createRegion(rootName, "bridge");
     Log.getLogWriter().info("Created root region " + rootName);
     createSubRegions(rootRegion, breadth, depth, "Region");
   }

   // start the bridge server
   BridgeHelper.startBridgeServer("bridge");
}
 

public void testInvalidate() throws Exception {
  SerializableCallable invalid = new SerializableCallable() {
    @Override
    public Object call() throws Exception {
      Region<Integer, MyObject> r = getCache().getRegion("clienttest");

      r.put(1, new MyObject(1, "invalidate"));
      r.invalidate(1);

      File f = new File(getDiskDirs()[0], "client-invalidate.snapshot");
      r.getSnapshotService().save(f, SnapshotFormat.GEMFIRE);
      r.getSnapshotService().load(f, SnapshotFormat.GEMFIRE);
      
      return null;
    }
  };
  
  Host.getHost(0).getVM(3).invoke(invalid);
  
  assertTrue(region.containsKey(1));
  assertFalse(region.containsValueForKey(1));
  assertNull(region.get(1));
}
 

@Test
public void testIsRegionPersistentWhenDataPolicyIsPartition() {
  final Region mockRegion = mockContext.mock(Region.class, "Region");
  final RegionAttributes mockRegionAttributes = mockContext.mock(RegionAttributes.class, "RegionAttributes");

  mockContext.checking(new Expectations() {{
    oneOf(mockRegion).getAttributes();
    will(returnValue(mockRegionAttributes));
    oneOf(mockRegionAttributes).getDataPolicy();
    will(returnValue(DataPolicy.PARTITION));
  }});

  final DescribeDiskStoreFunction function = createDescribeDiskStoreFunction(null);

  assertFalse(function.isPersistent(mockRegion));
}
 

private static String getLeaderRegionPath(String regionPath,
    RegionAttributes regionAttributes, Cache cache) {
  String colocated;
  while(regionAttributes.getPartitionAttributes() != null 
      && (colocated = regionAttributes.getPartitionAttributes().getColocatedWith()) != null) {
    // Do not waitOnInitialization() for PR
    GemFireCacheImpl gfc = (GemFireCacheImpl)cache;
    Region colocatedRegion = gfc.getPartitionedRegion(colocated, false);
    if(colocatedRegion == null) {
      Assert.fail("Could not find parent region " + colocated + " for " + regionPath);
    }
    regionAttributes = colocatedRegion.getAttributes();
    regionPath = colocatedRegion.getFullPath();
  }
  return regionPath;
}
 

private void createPartitionedRegionWithNullValues() {
  Cache cache = CacheUtils.getCache();
  PartitionAttributesFactory prAttFactory = new PartitionAttributesFactory();
  AttributesFactory attributesFactory = new AttributesFactory();
  attributesFactory.setPartitionAttributes(prAttFactory.create());
  RegionAttributes regionAttributes = attributesFactory.create();
  Region region = cache.createRegion(regionName, regionAttributes);

  for (int i = 1; i <= numElem; i++) {
    Portfolio obj = new Portfolio(i);
    region.put(i, obj);
    if (i%(numElem/5) == 0) obj.status = null;
    region.put(i + numElem, obj);
    System.out.println(obj);
  }
}
 

public void testExport() throws Exception {
  int rcount = 0;
  for (final RegionType rt : RegionType.persistentValues()) {
    for (final SerializationType st : SerializationType.offlineValues()) {
      Region<Integer, MyObject> region = rgen.createRegion(cache, ds.getName(), rt, "test" + rcount++);
      final Map<Integer, MyObject> expected = createExpected(st, 1000);
      
      region.putAll(expected);
      cache.close();
      
      DiskStoreImpl.exportOfflineSnapshot(ds.getName(), new File[] { new File(".") }, new File("."));
      checkSnapshotEntries(expected, ds.getName(), region.getName());
      
      reset();
    }
  }
}
 

public void test_GatewaySenderIdAndAsyncEventId() {
    cache = new CacheFactory().create();
    AttributesFactory factory = new AttributesFactory();
    factory.addGatewaySenderId("ln");
    factory.addGatewaySenderId("ny");
    factory.addAsyncEventQueueId("Async_LN");
    RegionAttributes attrs = factory.create();
    
    Set<String> senderIds = new HashSet<String>();
    senderIds.add("ln");
    senderIds.add("ny");
    Set<String> attrsSenderIds = attrs.getGatewaySenderIds();
    assertEquals(senderIds, attrsSenderIds);
    Region r = cache.createRegion("Customer", attrs);
    assertEquals(senderIds, ((LocalRegion)r).getGatewaySenderIds());
}
 

public static void createEntries()
{
  try {
    Region r = CacheServerTestUtil.getCache().getRegion("/"+ REGION_NAME);
    assertNotNull(r);

    if (!r.containsKey("key-1")) {
      r.create("key-1", "key-1");
    }
    if (!r.containsKey("key-6")) {
      r.create("key-6", "key-6");
    }
    // Verify that no invalidates occurred to this region
    assertEquals(r.getEntry("key-1").getValue(), "key-1");
    assertEquals(r.getEntry("key-6").getValue(), "key-6");
  }
  catch (Exception ex) {
    fail("failed while createEntries()", ex);
  }
}
 

@Override
public void close() {
  Region r = getRegion();
  if (r != null && !r.isDestroyed()) {
    try {
      r.close();
    } catch (RegionDestroyedException e) {
    }
  }
}
 

/**
 * Task used for client Q conflation tests. The test currently uses the
 * product test hooks for the client Q conflation settings. Initializes the
 * test region in the cache client VM and registers interest.
 */
public static void initCacheClientWithConflation() {
  synchronized (HAClientQueue.class) {
    if (CacheHelper.getCache() == null) { // first thread

      // create the cache and region
      Cache cache = CacheHelper.createCache(ConfigPrms.getCacheConfig());
      
      int numOfRegion = TestConfig.tab().intAt(
          hct.ha.HAClientQueuePrms.numberOfRegions, 1);
      for (int i = 0; i < numOfRegion; i++) {
        Region region = RegionHelper.createRegion(REGION_NAME + i, ConfigPrms
            .getRegionConfig());


        mypool = ClientHelper.getPool(region);
        
        Log.getLogWriter().info(
            "The primary server endpoint is " + mypool.getPrimaryName());

        // register interest in one thread to avoid possible performance hit
        try {
          region.registerInterest("ALL_KEYS");
        }
        catch (CacheWriterException e) {
          throw new TestException(TestHelper.getStackTrace(e));
        }
      }

    }
  }
}
 

public void test010PUTDMLSupport() {
  Region<Integer, String> r = createRegion(getName());
  SortedOplogStatistics stats = HDFSRegionDirector.getInstance().getHdfsRegionStats("/" + getName());
  assertEquals(0, stats.getRead().getCount());
  for (int i=0; i<100; i++) {
    r.put(i, "value"+i);
  }
  assertEquals(100, stats.getRead().getCount());
  sleep(r.getFullPath());
  clearBackingCHM(r);
  LocalRegion lr = (LocalRegion) r;
  for (int i=0; i<200; i++) {
    EntryEventImpl ev = lr.newPutEntryEvent(i, "value"+i, null);
    lr.validatedPut(ev, System.currentTimeMillis());
  }
  // verify that read count on HDFS does not change
  assertEquals(100, stats.getRead().getCount());
  sleep(r.getFullPath());
  clearBackingCHM(r);
  for (int i=0; i<200; i++) {
    assertEquals("value"+i, r.get(i));
  }
  if (getBatchTimeInterval() > 1000) {
    // reads from async queue
    assertEquals(100, stats.getRead().getCount());
  } else {
    assertEquals(300, stats.getRead().getCount());
  }
}
 

/**
 * Generates events having specific values of threadId and sequenceId, via
 * clearRegionOperation through connection object
 * 
 * @throws Exception -
 *           thrown if any problem occurs in clearRegionOperation
 */
public static void generateEventsByClearRegionOperation() throws Exception
{
  Connection connection = pool.acquireConnection();
  String regionName = Region.SEPARATOR + REGION_NAME;
  ServerRegionProxy srp = new ServerRegionProxy(regionName, pool);

  for (int i = 0; i < eventIds.length; i++) {
    srp.clearOnForTestsOnly(connection, eventIds[i], null);
  }
  srp.clearOnForTestsOnly(connection, eventIdForLastKey, null);
}
 

/** Create region(s) in hydra.RegionPrms.names, using the hydra param
 *  PdxPrms.createProxyRegions to determine which regions specified in
 *  hdyra.RegionPrms.names to create.
 */
public static void createRegions() {
  List<String> regionConfigNamesList = TestConfig.tab().vecAt(RegionPrms.names);
  for (String regionConfigName: regionConfigNamesList) {
    if (shouldCreateRegion(regionConfigName)) {
      Log.getLogWriter().info("Creating region with config name " + regionConfigName);
      Region aRegion = RegionHelper.createRegion(regionConfigName);
      Log.getLogWriter().info("Done creating region with config name " + regionConfigName + ", region name is " + aRegion.getFullPath());
    }
  }
  if (CommandPrms.getCreateSubregions()) {
    createSubregions();
  }
  Log.getLogWriter().info(TestHelper.regionHierarchyToString());
}
 

private void putProperties(Region<Object, Object> region, List<ExpectedValues> expected) {
  String key = "keyProperties";
  Properties value = new Properties();
  value.put("1", "string 1");
  value.put("2", "string 2");
  region.put(key, value);
  expected.add(new ExpectedValues(value, 40, "java.util.Properties", -1, getMemoryAddress(region, key), 1, 0, false, true));
}
 

protected void localDestroyPR(VM vm0) {
  SerializableRunnable destroyPR = new SerializableRunnable("destroy pr") {

    public void run() {
      Cache cache = getCache();
      Region region = cache.getRegion(PR_REGION_NAME);
      region.localDestroyRegion();
    }
  };
  vm0.invoke(destroyPR);
}
 

/**
 *
 * @param regionNames
 */
private synchronized static void InitializeRegions(String[] regionNames) {
  for (String regionName : regionNames) {
    Region aRegion = RegionHelper.createRegion(regionName);
    testInstance.theRegions.add(aRegion);
    Log.getLogWriter().info("SerialCompressionTest.InitializeRegions regionName=" + aRegion.getName() +
                            " Compressor=" + aRegion.getAttributes().getCompressor().getClass().getName());
  }

  testInstance.logRegionHierarchy();
}
 

/**
 * Log the local size of the PR data store
 */
public synchronized static void HydraTask_logLocalSize() {
  //Set<Region<?, ?>> regionSet = theCache.rootRegions();
  Set<Region<?, ?>> regionSet = ((FixedPartitioningTest)testInstance)
      .getTestRegions();
  for (Region aRegion : regionSet) {
    Log.getLogWriter().info(
        "Number of entries in this data store: "
            + ParRegUtil.getLocalSize(aRegion));
  }
}
 

public static Collection<?> executeFunctionOnRegionWithArgs(Region<?, ?> rgn,
    Object args, String functionId, ResultCollector<?, ?> rc,
    boolean enableStreaming, boolean isPossibleDuplicate, boolean isHA,
    MemberMappedArgument mma, Set<Object> routingObjects,
    AbstractGemFireResultSet rs) throws StandardException {
  final GfxdExecution exec = onRegion(rgn);
  return (Collection<?>)executeFunction(exec, args, functionId, functionId,
      GfxdConstants.HA_NUM_RETRIES, rc, enableStreaming, isPossibleDuplicate,
      isHA, mma, routingObjects, false, rs);
}
 

public static void destroyRegion1()
{
  try {
    Cache cache = new BridgeWriterMiscDUnitTest("temp").getCache();
    Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1);
    assertNotNull(r1);
    r1.destroyRegion();
  }
  catch (Exception e) {
    e.printStackTrace();
    fail("Test failed due to Exception during closeBothRegions", e);
  }
}
 

/** Verify containsValueForKey for the given region and key.
 *
 * @param aRegion The region to verify.
 * @param key The key in aRegion to verify.
 * @param expected The expected value of containsKey()
 *
 * @throws TestException if containsValueforKey() has the wrong value
 */
public static void verifyContainsValueForKey(Region aRegion, Object key, boolean expected) {
   boolean containsValueForKey = aRegion.containsValueForKey(key);
   if (containsValueForKey != expected) {
      RegionAttributes attr = aRegion.getAttributes();
      throw new TestException("Expected containsValueForKey() for " + key + " to be " + expected + 
                " in " + aRegion.getFullPath() + ", but it is " + containsValueForKey);
   }
}
 

/**
 * Initializes the test region in the cache client VM and registers interest
 * only for non delta's.
 */
public static void initCacheClientOld() {
  synchronized (DeltaPropagation.class) {
    if (CacheHelper.getCache() == null) { // first thread
      // create the cache and region
      CacheHelper.createCache(ConfigPrms.getCacheConfig());

      int numOfRegion = TestConfig.tab().intAt(
          delta.DeltaPropagationPrms.numberOfRegions, 1);
      for (int i = 0; i < numOfRegion; i++) {
        Region region = RegionHelper.createRegion(REGION_NAME + i, ConfigPrms
            .getRegionConfig());

        // set the loader and writer statics (same for all regions)
        mypool = ClientHelper.getPool(region);

        try {
          region.registerInterestRegex(".*_n_d_o");
          region.registerInterest("last_key");
          
        }
        catch (CacheWriterException e) {
          throw new TestException(TestHelper.getStackTrace(e));
        }
      }
    }
  }
}
 

/**
 * Create a Schema in the store in given transaction context.
 * 
 * @param schemaName
 *          the name of the schema
 * @param tc
 *          the {@link TransactionManager} for this operation
 */
public Region<?, ?> createSchemaRegion(String schemaName,
    TransactionController tc) throws StandardException {
  Region<?, ?> schemaRegion = null;
  boolean locked = false;
  final GemFireTransaction tran = (GemFireTransaction)tc;
  LockingPolicy locking = null;
  if (!tran.skipLocks(schemaName, null)) {
    locking = new GFContainerLocking(
        new DefaultGfxdLockable(schemaName, null), false, null);
    // take distributed write lock before create
    // the lock will be released when transaction ends
    locked = locking.lockContainer(tran, null, true, true);
    // check if someone else created the schema region in the meantime
    if ((schemaRegion = this.gemFireCache.getRegion(schemaName)) != null) {
      return schemaRegion;
    }
  }
  AttributesFactory<?, ?> af = new AttributesFactory<Object, Object>();
  af.setScope(Scope.DISTRIBUTED_NO_ACK);
  af.setDataPolicy(DataPolicy.EMPTY);
  af.setConcurrencyChecksEnabled(false);
  try {
    schemaRegion = this.gemFireCache.createRegion(schemaName, af.create());
  } catch (RegionExistsException ex) {
    // [sumedh] A genuine case should be caught at the derby level
    // while other case where it can arise is receiving a GfxdDDLMessage
    // for a schema creation that has already been replayed using
    // the hidden _DDL_STMTS_REGION.
    final LogWriterI18n logger = this.gemFireCache.getLoggerI18n();
    if (logger.finerEnabled()) {
      logger.finer("createSchemaRegion: region for schema '" + schemaName
          + "' already exists", ex);
    }
    if (locked) {
      locking.unlockContainer(tran, null);
    }
  }
  return schemaRegion;
}
 

/**
 * do three puts on key-1
 * 
 */
public static void putValue1()
{
  try {
    Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME);
    r1.put("key-1", "value-1");
  }
  catch (Exception ex) {
    ex.printStackTrace();
    fail("failed while region.put()", ex);
  }
}