/*
 * 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.
 */
package org.apache.geode.internal.cache.wan;

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.distributed.ConfigurationProperties.CONSERVE_SOCKETS;
import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_CIPHERS;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_ENABLED;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_KEYSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_KEYSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_KEYSTORE_TYPE;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_REQUIRE_AUTHENTICATION;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_TRUSTSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_TRUSTSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_HTTP_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_START;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.OFF_HEAP_MEMORY_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.Host.getHost;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.apache.geode.test.util.ResourceUtils.createTempFileFromResource;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.management.ObjectName;

import org.apache.commons.io.FileUtils;
import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.NotNull;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.experimental.categories.Category;

import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.TransactionException;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.ConnectionStats;
import org.apache.geode.cache.client.internal.LocatorDiscoveryCallbackAdapter;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.client.internal.locator.wan.LocatorMembershipListener;
import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.wan.GatewayEventFilter;
import org.apache.geode.cache.wan.GatewayQueueEvent;
import org.apache.geode.cache.wan.GatewayReceiver;
import org.apache.geode.cache.wan.GatewayReceiverFactory;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.cache.wan.GatewaySender.OrderPolicy;
import org.apache.geode.cache.wan.GatewaySenderFactory;
import org.apache.geode.cache.wan.GatewayTransportFilter;
import org.apache.geode.cache.wan.internal.GatewaySenderEventRemoteDispatcher;
import org.apache.geode.cache30.CacheTestCase;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.Locator;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.admin.remote.DistributionLocatorId;
import org.apache.geode.internal.cache.BucketRegion;
import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.CustomerIDPartitionResolver;
import org.apache.geode.internal.cache.ForceReattemptException;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCacheBuilder;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PoolStats;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.execute.data.CustId;
import org.apache.geode.internal.cache.execute.data.Customer;
import org.apache.geode.internal.cache.execute.data.Order;
import org.apache.geode.internal.cache.execute.data.OrderId;
import org.apache.geode.internal.cache.execute.data.Shipment;
import org.apache.geode.internal.cache.execute.data.ShipmentId;
import org.apache.geode.internal.cache.partitioned.BecomePrimaryBucketMessage;
import org.apache.geode.internal.cache.partitioned.BecomePrimaryBucketMessage.BecomePrimaryBucketResponse;
import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException;
import org.apache.geode.internal.cache.tier.sockets.CacheServerStats;
import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil;
import org.apache.geode.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderEventProcessor;
import org.apache.geode.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderQueue;
import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderEventProcessor;
import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue;
import org.apache.geode.internal.cache.wan.serial.ConcurrentSerialGatewaySenderEventProcessor;
import org.apache.geode.internal.cache.wan.serial.SerialGatewaySenderQueue;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.management.AsyncEventQueueMXBean;
import org.apache.geode.management.DistributedSystemMXBean;
import org.apache.geode.management.GatewayReceiverMXBean;
import org.apache.geode.management.GatewaySenderMXBean;
import org.apache.geode.management.MBeanUtil;
import org.apache.geode.management.ManagementService;
import org.apache.geode.management.RegionMXBean;
import org.apache.geode.management.internal.SystemManagementService;
import org.apache.geode.pdx.SimpleClass;
import org.apache.geode.pdx.SimpleClass1;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.DistributedTestCase;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.junit.categories.WanTest;

@SuppressWarnings("deprecation")
@Category({WanTest.class})
public class WANTestBase extends DistributedTestCase {

  protected static Cache cache;
  protected static Region<?, ?> region;

  protected static PartitionedRegion customerRegion;
  protected static PartitionedRegion orderRegion;
  protected static PartitionedRegion shipmentRegion;

  protected static final String customerRegionName = "CUSTOMER";
  protected static final String orderRegionName = "ORDER";
  protected static final String shipmentRegionName = "SHIPMENT";

  protected static VM vm0;
  protected static VM vm1;
  protected static VM vm2;
  protected static VM vm3;
  protected static VM vm4;
  protected static VM vm5;
  protected static VM vm6;
  protected static VM vm7;

  protected static QueueListener<Object, Object> listener1;
  protected static QueueListener<Object, Object> listener2;

  protected static AsyncEventListener eventListener1;
  protected static AsyncEventListener eventListener2;

  private static final long MAX_WAIT = 10000;

  protected static GatewayEventFilter eventFilter;

  protected static List<Integer> dispatcherThreads = new ArrayList<>(Arrays.asList(1, 3, 5));
  // this will be set for each test method run with one of the values from above list
  protected static int numDispatcherThreadsForTheRun = 1;

  private static final Logger logger = LogService.getLogger();

  @BeforeClass
  public static void beforeClassWANTestBase() {
    vm0 = getHost(0).getVM(0);
    vm1 = getHost(0).getVM(1);
    vm2 = getHost(0).getVM(2);
    vm3 = getHost(0).getVM(3);
    vm4 = getHost(0).getVM(4);
    vm5 = getHost(0).getVM(5);
    vm6 = getHost(0).getVM(6);
    vm7 = getHost(0).getVM(7);
  }

  @Before
  public void setUpWANTestBase() throws Exception {
    shuffleNumDispatcherThreads();
    Invoke.invokeInEveryVM(() -> setNumDispatcherThreadsForTheRun(dispatcherThreads.get(0)));
    addIgnoredException("Connection refused");
    addIgnoredException("Software caused connection abort");
    addIgnoredException("Connection reset");
    postSetUpWANTestBase();
  }

  protected void postSetUpWANTestBase() throws Exception {
    // nothing
  }

  public static void shuffleNumDispatcherThreads() {
    Collections.shuffle(dispatcherThreads);
  }

  public static void setNumDispatcherThreadsForTheRun(int numThreads) {
    numDispatcherThreadsForTheRun = numThreads;
  }

  public static void stopOldLocator() {
    if (Locator.hasLocator()) {
      Locator.getLocator().stop();
    }
  }

  public static void createLocator(int dsId, int port, Set<String> localLocatorsList,
      Set<String> remoteLocatorsList) {
    WANTestBase test = new WANTestBase();
    Properties props = test.getDistributedSystemProperties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId);
    StringBuilder localLocatorBuffer = new StringBuilder(localLocatorsList.toString());
    localLocatorBuffer.deleteCharAt(0);
    localLocatorBuffer.deleteCharAt(localLocatorBuffer.lastIndexOf("]"));
    String localLocator = localLocatorBuffer.toString();
    localLocator = localLocator.replace(" ", "");

    props.setProperty(LOCATORS, localLocator);
    props.setProperty(START_LOCATOR,
        "localhost[" + port + "],server=true,peer=true,hostname-for-clients=localhost");
    StringBuilder remoteLocatorBuffer = new StringBuilder(remoteLocatorsList.toString());
    remoteLocatorBuffer.deleteCharAt(0);
    remoteLocatorBuffer.deleteCharAt(remoteLocatorBuffer.lastIndexOf("]"));
    String remoteLocator = remoteLocatorBuffer.toString();
    remoteLocator = remoteLocator.replace(" ", "");
    props.setProperty(REMOTE_LOCATORS, remoteLocator);
    test.startLocatorDistributedSystem(props);
  }

  private void startLocator(int dsId, int locatorPort, int startLocatorPort, int remoteLocPort,
      boolean startServerLocator) {
    Properties props = getDistributedSystemProperties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId);
    props.setProperty(LOCATORS, "localhost[" + locatorPort + "]");
    props.setProperty(START_LOCATOR, "localhost[" + startLocatorPort + "],server="
        + startServerLocator + ",peer=true,hostname-for-clients=localhost");
    if (remoteLocPort != -1) {
      props.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocPort + "]");
    }
    startLocatorDistributedSystem(props);
  }

  private void startLocatorDistributedSystem(Properties props) {
    // Start start the locator with a LOCATOR_DM_TYPE and not a NORMAL_DM_TYPE
    System.setProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE, "true");
    try {
      getSystem(props);
    } finally {
      System.clearProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE);
    }
  }

  public static Integer createFirstLocatorWithDSId(int dsId) {
    stopOldLocator();
    WANTestBase test = new WANTestBase();
    int port = getRandomAvailableTCPPort();
    test.startLocator(dsId, port, port, -1, true);
    return port;
  }

  public static Integer createFirstPeerLocator(int dsId) {
    stopOldLocator();
    WANTestBase test = new WANTestBase();
    int port = getRandomAvailableTCPPort();
    test.startLocator(dsId, port, port, -1, false);
    return port;
  }

  public static Integer createSecondLocator(int dsId, int locatorPort) {
    stopOldLocator();
    WANTestBase test = new WANTestBase();
    int port = getRandomAvailableTCPPort();
    test.startLocator(dsId, locatorPort, port, -1, true);
    return port;
  }

  public static Integer createSecondPeerLocator(int dsId, int locatorPort) {
    stopOldLocator();
    WANTestBase test = new WANTestBase();
    int port = getRandomAvailableTCPPort();
    test.startLocator(dsId, locatorPort, port, -1, false);
    return port;
  }

  public static Integer createFirstRemoteLocator(int dsId, int remoteLocPort) {
    stopOldLocator();
    WANTestBase test = new WANTestBase();
    int port = getRandomAvailableTCPPort();
    test.startLocator(dsId, port, port, remoteLocPort, true);
    return port;
  }

  public static void bringBackLocatorOnOldPort(int dsId, int remoteLocPort, int oldPort) {
    WANTestBase test = new WANTestBase();
    test.startLocator(dsId, oldPort, oldPort, remoteLocPort, true);
  }


  public static Integer createFirstRemotePeerLocator(int dsId, int remoteLocPort) {
    stopOldLocator();
    WANTestBase test = new WANTestBase();
    int port = getRandomAvailableTCPPort();
    test.startLocator(dsId, port, port, remoteLocPort, false);
    return port;
  }

  public static Integer createSecondRemoteLocator(int dsId, int localPort, int remoteLocPort) {
    stopOldLocator();
    WANTestBase test = new WANTestBase();
    int port = getRandomAvailableTCPPort();
    test.startLocator(dsId, localPort, port, remoteLocPort, true);
    return port;
  }

  public static Integer createSecondRemoteLocatorWithAPI(int dsId, int localPort, int remoteLocPort,
      String hostnameForClients) throws IOException {
    stopOldLocator();
    WANTestBase test = new WANTestBase();
    Properties props = test.getDistributedSystemProperties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId);
    props.setProperty(LOCATORS, "localhost[" + localPort + "]");
    props.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocPort + "]");
    Locator locator = Locator.startLocatorAndDS(0, null, null, props,
        true, true, hostnameForClients);
    return locator.getPort();
  }

  public static Integer createSecondRemotePeerLocator(int dsId, int localPort, int remoteLocPort) {
    stopOldLocator();
    WANTestBase test = new WANTestBase();
    int port = getRandomAvailableTCPPort();
    test.startLocator(dsId, localPort, port, remoteLocPort, false);
    return port;
  }

  public static int createReceiverInSecuredCache() {
    GatewayReceiverFactory fact = WANTestBase.cache.createGatewayReceiverFactory();
    int port = getRandomAvailableTCPPort();
    fact.setStartPort(port);
    fact.setEndPort(port);
    fact.setManualStart(true);
    GatewayReceiver receiver = fact.create();
    try {
      receiver.start();
    } catch (IOException e) {
      e.printStackTrace();
      fail("Failed to start GatewayReceiver on port " + port, e);
    }
    return port;
  }

  public static void createReplicatedRegion(String regionName, String senderIds, Boolean offHeap) {
    IgnoredException exp =
        addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        addIgnoredException(InterruptedException.class.getName());
    IgnoredException exp2 =
        addIgnoredException(GatewaySenderException.class.getName());
    try {
      RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.REPLICATE);
      if (senderIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String senderId = tokenizer.nextToken();
          fact.addGatewaySenderId(senderId);
        }
      }

      fact.setOffHeap(offHeap);
      fact.create(regionName);
    } finally {
      exp.remove();
      exp1.remove();
      exp2.remove();
    }
  }

  public static void createReplicatedProxyRegion(String regionName, String senderIds,
      Boolean offHeap) {
    IgnoredException exp =
        addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        addIgnoredException(InterruptedException.class.getName());
    IgnoredException exp2 =
        addIgnoredException(GatewaySenderException.class.getName());
    try {
      RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.REPLICATE_PROXY);
      if (senderIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String senderId = tokenizer.nextToken();
          fact.addGatewaySenderId(senderId);
        }
      }

      fact.setOffHeap(offHeap);
      fact.create(regionName);
    } finally {
      exp.remove();
      exp1.remove();
      exp2.remove();
    }
  }

  public static void createNormalRegion(String regionName, String senderIds) {
    RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.LOCAL);
    if (senderIds != null) {
      StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
      while (tokenizer.hasMoreTokens()) {
        String senderId = tokenizer.nextToken();
        fact.addGatewaySenderId(senderId);
      }
    }

    fact.create(regionName);
  }

  public static void createPersistentReplicatedRegion(String regionName, String senderIds,
      Boolean offHeap) {
    RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.REPLICATE_PERSISTENT);
    if (senderIds != null) {
      StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
      while (tokenizer.hasMoreTokens()) {
        String senderId = tokenizer.nextToken();
        fact.addGatewaySenderId(senderId);
      }
    }

    fact.setOffHeap(offHeap);
    fact.create(regionName);
  }

  public static void createReplicatedRegionWithAsyncEventQueue(String regionName,
      String asyncQueueIds, Boolean offHeap) {
    IgnoredException exp1 =
        addIgnoredException(ForceReattemptException.class.getName());
    try {
      RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.REPLICATE);
      if (asyncQueueIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(asyncQueueIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String asyncQueueId = tokenizer.nextToken();
          fact.addAsyncEventQueueId(asyncQueueId);
        }
      }

      fact.setOffHeap(offHeap);
      fact.create(regionName);
    } finally {
      exp1.remove();
    }
  }

  public static void createReplicatedRegionWithSenderAndAsyncEventQueue(String regionName,
      String senderIds, String asyncChannelId, Boolean offHeap) {
    IgnoredException exp =
        addIgnoredException(ForceReattemptException.class.getName());
    try {
      RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.REPLICATE);
      if (senderIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String senderId = tokenizer.nextToken();
          fact.addGatewaySenderId(senderId);
        }
      }

      fact.setOffHeap(offHeap);
      fact.addAsyncEventQueueId(asyncChannelId);
      fact.create(regionName);
    } finally {
      exp.remove();
    }
  }

  public static void createReplicatedRegion(String regionName, String senderIds, Scope scope,
      DataPolicy policy, Boolean offHeap) {
    createReplicatedRegion(regionName, senderIds, scope, policy, offHeap, false, true);
  }

  public static void createReplicatedRegion(String regionName, String senderIds, Scope scope,
      DataPolicy policy, Boolean offHeap, boolean statisticsEnabled,
      boolean concurrencyChecksEnabled) {
    RegionFactory<?, ?> fact = cache.createRegionFactory();
    if (senderIds != null) {
      StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
      while (tokenizer.hasMoreTokens()) {
        String senderId = tokenizer.nextToken();
        fact.addGatewaySenderId(senderId);
      }
    }

    fact.setDataPolicy(policy);
    fact.setScope(scope);
    fact.setOffHeap(offHeap);
    fact.setStatisticsEnabled(statisticsEnabled);
    fact.setConcurrencyChecksEnabled(concurrencyChecksEnabled);
    fact.create(regionName);
  }

  public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent,
      String diskStoreName, boolean isDiskSynchronous) {

    if (diskStoreName != null) {
      File directory = new File(
          asyncChannelId + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum());
      directory.mkdir();
      File[] dirs1 = new File[] {directory};
      DiskStoreFactory dsf = cache.createDiskStoreFactory();
      dsf.setDiskDirs(dirs1);
      dsf.create(diskStoreName);
    }

    AsyncEventListener asyncEventListener = new MyAsyncEventListener();

    AsyncEventQueueFactory factory = cache.createAsyncEventQueueFactory();
    factory.setBatchSize(batchSize);
    factory.setPersistent(isPersistent);
    factory.setDiskStoreName(diskStoreName);
    factory.setDiskSynchronous(isDiskSynchronous);
    factory.setBatchConflationEnabled(isConflation);
    factory.setMaximumQueueMemory(maxMemory);
    factory.setParallel(isParallel);
    // set dispatcher threads
    factory.setDispatcherThreads(numDispatcherThreadsForTheRun);
    factory.create(asyncChannelId, asyncEventListener);
  }

  public static void createPartitionedRegion(String regionName, String senderIds,
      Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap) {
    createPartitionedRegion(regionName, senderIds, redundantCopies, totalNumBuckets, offHeap,
        RegionShortcut.PARTITION);
  }

  public static void createPartitionedRegion(String regionName, String senderIds,
      Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap, RegionShortcut shortcut) {
    createPartitionedRegion(regionName, senderIds, redundantCopies, totalNumBuckets, offHeap,
        shortcut, false, true);
  }

  public static void createPartitionedRegion(String regionName, String senderIds,
      Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap, RegionShortcut shortcut,
      boolean statisticsEnabled, boolean concurrencyChecksEnabled) {
    IgnoredException exp =
        addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        addIgnoredException(PartitionOfflineException.class.getName());
    try {
      RegionFactory<?, ?> fact = cache.createRegionFactory(shortcut);
      if (senderIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String senderId = tokenizer.nextToken();
          fact.addGatewaySenderId(senderId);
        }
      }
      PartitionAttributesFactory<?, ?> pfact = new PartitionAttributesFactory<>();
      pfact.setTotalNumBuckets(totalNumBuckets);
      pfact.setRedundantCopies(redundantCopies);
      pfact.setRecoveryDelay(0);
      fact.setPartitionAttributes(pfact.create());
      fact.setOffHeap(offHeap);
      fact.setStatisticsEnabled(statisticsEnabled);
      fact.setConcurrencyChecksEnabled(concurrencyChecksEnabled);
      fact.create(regionName);
    } finally {
      exp.remove();
      exp1.remove();
    }
  }

  // TODO:OFFHEAP: add offheap flavor
  public static void createPartitionedRegionWithPersistence(String regionName, String senderIds,
      Integer redundantCopies, Integer totalNumBuckets) {
    IgnoredException exp =
        addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        addIgnoredException(PartitionOfflineException.class.getName());
    try {
      RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.PARTITION_PERSISTENT);
      if (senderIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String senderId = tokenizer.nextToken();
          fact.addGatewaySenderId(senderId);
        }
      }
      PartitionAttributesFactory<?, ?> pfact = new PartitionAttributesFactory<>();
      pfact.setTotalNumBuckets(totalNumBuckets);
      pfact.setRedundantCopies(redundantCopies);
      pfact.setRecoveryDelay(0);
      fact.setPartitionAttributes(pfact.create());
      fact.create(regionName);
    } finally {
      exp.remove();
      exp1.remove();
    }
  }

  public static void createColocatedPartitionedRegion(String regionName, String senderIds,
      Integer redundantCopies, Integer totalNumBuckets, String colocatedWith) {
    IgnoredException exp =
        addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        addIgnoredException(PartitionOfflineException.class.getName());
    try {
      RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.PARTITION);
      if (senderIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String senderId = tokenizer.nextToken();
          fact.addGatewaySenderId(senderId);
        }
      }

      PartitionAttributesFactory<?, ?> pfact = new PartitionAttributesFactory<>();
      pfact.setTotalNumBuckets(totalNumBuckets);
      pfact.setRedundantCopies(redundantCopies);
      pfact.setRecoveryDelay(0);
      pfact.setColocatedWith(colocatedWith);
      fact.setPartitionAttributes(pfact.create());
      fact.create(regionName);
    } finally {
      exp.remove();
      exp1.remove();
    }
  }

  public static void addSenderThroughAttributesMutator(String regionName, String senderIds) {
    final Region<?, ?> r = cache.getRegion(SEPARATOR + regionName);
    AttributesMutator<?, ?> mutator = r.getAttributesMutator();
    mutator.addGatewaySenderId(senderIds);
  }

  public static void addAsyncEventQueueThroughAttributesMutator(String regionName, String queueId) {
    final Region<?, ?> r = cache.getRegion(SEPARATOR + regionName);
    AttributesMutator<?, ?> mutator = r.getAttributesMutator();
    mutator.addAsyncEventQueueId(queueId);
  }

  public static void createPartitionedRegionAsAccessor(String regionName, String senderIds,
      Integer redundantCopies, Integer totalNumBuckets) {
    RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.PARTITION_PROXY);
    if (senderIds != null) {
      StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
      while (tokenizer.hasMoreTokens()) {
        String senderId = tokenizer.nextToken();
        fact.addGatewaySenderId(senderId);
      }
    }
    PartitionAttributesFactory<?, ?> pfact = new PartitionAttributesFactory<>();
    pfact.setTotalNumBuckets(totalNumBuckets);
    pfact.setRedundantCopies(redundantCopies);
    fact.setPartitionAttributes(pfact.create());
    fact.create(regionName);
  }

  public static void createPartitionedRegionWithSerialParallelSenderIds(String regionName,
      String serialSenderIds, String parallelSenderIds, String colocatedWith, Boolean offHeap) {
    RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.PARTITION);
    if (serialSenderIds != null) {
      StringTokenizer tokenizer = new StringTokenizer(serialSenderIds, ",");
      while (tokenizer.hasMoreTokens()) {
        String senderId = tokenizer.nextToken();
        fact.addGatewaySenderId(senderId);
      }
    }
    if (parallelSenderIds != null) {
      StringTokenizer tokenizer = new StringTokenizer(parallelSenderIds, ",");
      while (tokenizer.hasMoreTokens()) {
        String senderId = tokenizer.nextToken();
        fact.addGatewaySenderId(senderId);
      }
    }
    PartitionAttributesFactory<?, ?> pfact = new PartitionAttributesFactory<>();
    pfact.setColocatedWith(colocatedWith);
    fact.setPartitionAttributes(pfact.create());
    fact.setOffHeap(offHeap);
    fact.create(regionName);
  }

  public static void createPersistentPartitionedRegion(String regionName, String senderIds,
      Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap) {

    IgnoredException exp =
        addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        addIgnoredException(PartitionOfflineException.class.getName());
    try {

      RegionFactory<?, ?> fact = cache.createRegionFactory(RegionShortcut.PARTITION_PERSISTENT);
      if (senderIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String senderId = tokenizer.nextToken();
          fact.addGatewaySenderId(senderId);
        }
      }
      PartitionAttributesFactory<?, ?> pfact = new PartitionAttributesFactory<>();
      pfact.setTotalNumBuckets(totalNumBuckets);
      pfact.setRedundantCopies(redundantCopies);
      fact.setPartitionAttributes(pfact.create());
      fact.setOffHeap(offHeap);
      fact.create(regionName);
    } finally {
      exp.remove();
      exp1.remove();
    }
  }

  public static void createCustomerOrderShipmentPartitionedRegion(String senderIds,
      Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap) {
    createCustomerOrderShipmentPartitionedRegion(senderIds, redundantCopies, totalNumBuckets,
