/*
 * 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;

import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.OFF_HEAP_MEMORY_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.apache.geode.test.dunit.Assert.assertEquals;
import static org.apache.geode.test.dunit.Assert.assertTrue;
import static org.apache.geode.test.dunit.Assert.fail;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;

import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ClientRegionFactory;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.cache30.ClientServerTestCase;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.offheap.MemoryAllocatorImpl;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.DistributedTestUtils;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.junit.categories.ClientServerTest;

/**
 * Class <code>ClientServerGetAllDUnitTest</code> test client/server getAll.
 *
 * @since GemFire 5.7
 */
@Category({ClientServerTest.class})
public class ClientServerGetAllDUnitTest extends ClientServerTestCase {

  @Override
  public final void postTearDownCacheTestCase() throws Exception {
    disconnectAllFromDS();
  }

  @Test
  public void testGetAllFromServer() throws Exception {
    final Host host = Host.getHost(0);
    final VM server = host.getVM(0);
    final VM client = host.getVM(1);
    final String regionName = getUniqueName();
    final int serverPort = getRandomAvailableTCPPort();
    final String serverHost = NetworkUtils.getServerHostName(server.getHost());

    createBridgeServer(server, regionName, serverPort, false, false);

    createBridgeClient(client, regionName, serverHost, new int[] {serverPort});

    // Run getAll
    client.invoke(new CacheSerializableRunnable("Get all entries from server") {
      @Override
      public void run2() throws CacheException {
        // Build collection of keys
        Collection keys = new ArrayList();
        for (int i = 0; i < 5; i++) {
          keys.add("key-" + i);
        }

        keys.add(ClientServerTestCase.NON_EXISTENT_KEY); // this will not be load CacheLoader

        // Invoke getAll
        Region region = getRootRegion(regionName);
        Map result = region.getAll(keys);

        // Verify result size is correct
        assertEquals(6, result.size());

        // Verify the result contains each key,
        // and the value for each key is correct
        // (the server has a loader that returns the key as the value)
        for (final Object o : keys) {
          String key = (String) o;
          assertTrue(result.containsKey(key));
          Object value = result.get(key);
          if (!key.equals(ClientServerTestCase.NON_EXISTENT_KEY)) {
            assertEquals(key, value);
          } else {
            assertEquals(null, value);
          }
        }

        assertEquals(null, region.get(ClientServerTestCase.NON_EXISTENT_KEY));
      }
    });

    stopBridgeServer(server);
  }

  @Test
  public void testOffHeapGetAllFromServer() throws Exception {
    final Host host = Host.getHost(0);
    final VM server = host.getVM(0);
    final VM client = host.getVM(1);
    final String regionName = getUniqueName();
    final int serverPort = AvailablePortHelper.getRandomAvailableTCPPort();
    final String serverHost = NetworkUtils.getServerHostName(server.getHost());

    createBridgeServer(server, regionName, serverPort, false, false, true/* offheap */);

    createBridgeClient(client, regionName, serverHost, new int[] {serverPort});

    // Run getAll
    client.invoke(new CacheSerializableRunnable("Get all entries from server") {
      @Override
      public void run2() throws CacheException {
        // Build collection of keys
        Collection keys = new ArrayList();
        for (int i = 0; i < 5; i++) {
          keys.add("key-" + i);
        }

        keys.add(ClientServerTestCase.NON_EXISTENT_KEY); // this will not be load CacheLoader

        // Invoke getAll
        Region region = getRootRegion(regionName);
        Map result = region.getAll(keys);

        // Verify result size is correct
        assertEquals(6, result.size());

        // Verify the result contains each key,
        // and the value for each key is correct
        // (the server has a loader that returns the key as the value)
        for (final Object o : keys) {
          String key = (String) o;
          assertTrue(result.containsKey(key));
          Object value = result.get(key);
          if (!key.equals(ClientServerTestCase.NON_EXISTENT_KEY)) {
            assertEquals(key, value);
          } else {
            assertEquals(null, value);
          }
        }

        assertEquals(null, region.get(ClientServerTestCase.NON_EXISTENT_KEY));
      }
    });
    checkServerForOrphans(server, regionName);

    stopBridgeServer(server);
  }

  @Test
  public void testLargeOffHeapGetAllFromServer() throws Throwable {
    final Host host = Host.getHost(0);
    final VM server = host.getVM(0);
    final VM client = host.getVM(1);
    final String regionName = getUniqueName();
    final int serverPort = AvailablePortHelper.getRandomAvailableTCPPort();
    final String serverHost = NetworkUtils.getServerHostName(server.getHost());

    createBridgeServer(server, regionName, serverPort, false, false, true/* offheap */);

    createBridgeClient(client, regionName, serverHost, new int[] {serverPort}, true);

    final int VALUE_SIZE = 1024 * 2/* *1024 */;

    final int VALUE_COUNT = 100;

    client.invoke(new CacheSerializableRunnable("put entries on server") {
      @Override
      public void run2() throws CacheException {
        final byte[] VALUE = new byte[VALUE_SIZE];
        for (int i = 0; i < VALUE_SIZE; i++) {
          VALUE[i] = (byte) i;
        }
        Region region = getRootRegion(regionName);
        for (int i = 0; i < VALUE_COUNT; i++) {
          region.put("k" + i, new UnitTestValueHolder(VALUE));
        }
      }
    });

    CacheSerializableRunnable clientGetAll =
        new CacheSerializableRunnable("Get all entries from server") {
          @Override
          public void run2() throws CacheException {
            // Build collection of keys
            Collection keys = new ArrayList();
            for (int i = 0; i < VALUE_COUNT; i++) {
              keys.add("k" + i);
            }

            // Invoke getAll
            Region region = getRootRegion(regionName);
            final int GET_COUNT = 10/* 0 */;
            long start = System.currentTimeMillis();
            Map result = null;
            for (int i = 0; i < GET_COUNT; i++) {
              result = null; // allow gc to get rid of previous map before deserializing the next
                             // one
              result = region.getAll(keys);
            }
            long end = System.currentTimeMillis();
            long totalBytesRead = ((long) GET_COUNT * VALUE_COUNT * VALUE_SIZE);
            long elapsedMillis = (end - start);
            System.out.println("PERF: read " + totalBytesRead + " bytes in " + elapsedMillis
                + " millis. bps=" + (((double) totalBytesRead / elapsedMillis) * 1000));

            // Verify result size is correct
            assertEquals(VALUE_COUNT, result.size());

            final byte[] EXPECTED = new byte[VALUE_SIZE];
            for (int i = 0; i < VALUE_SIZE; i++) {
              EXPECTED[i] = (byte) i;
            }
            // Verify the result contains each key,
            // and the value for each key is correct
            // (the server has a loader that returns the key as the value)
            for (final Object o : keys) {
              String key = (String) o;
              assertTrue(result.containsKey(key));
              Object value = result.get(key);
              if (value instanceof UnitTestValueHolder) {
                Object v = ((UnitTestValueHolder) value).getValue();
                if (v instanceof byte[]) {
                  byte[] bytes = (byte[]) v;
                  if (bytes.length != VALUE_SIZE) {
                    fail("expected value for key " + key + " to be an array of size " + (VALUE_SIZE)
                        + " but it was: " + bytes.length);
                  }
                  if (!Arrays.equals(EXPECTED, bytes)) {
                    fail("expected bytes=" + Arrays.toString(bytes) + " to be expected="
                        + Arrays.toString(EXPECTED));
                  }
                } else {
                  fail("expected v for key " + key + " to be a byte array but it was: " + v);
                }
              } else {
                fail("expected value for key " + key + " to be a UnitTestValueHolder but it was: "
                    + value);
              }
            }
          }
        };
    // Run getAll
    {
      final int THREAD_COUNT = 4;
      AsyncInvocation<?>[] ais = new AsyncInvocation[THREAD_COUNT];
      for (int i = 0; i < THREAD_COUNT; i++) {
        ais[i] = client.invokeAsync(clientGetAll);
      }
      for (int i = 0; i < THREAD_COUNT; i++) {
        ais[i].getResult();
      }
    }

    server.invoke(new CacheSerializableRunnable("Dump OffHeap Stats") {
      @Override
      public void run2() throws CacheException {
        MemoryAllocatorImpl ma = MemoryAllocatorImpl.getAllocator();
        System.out.println("STATS: objects=" + ma.getStats().getObjects() + " usedMemory="
            + ma.getStats().getUsedMemory() + " reads=" + ma.getStats().getReads());
      }
    });

    checkServerForOrphans(server, regionName);

    stopBridgeServer(server);
  }

  @Override
  public Properties getDistributedSystemProperties() {
    Properties properties = super.getDistributedSystemProperties();
    properties.put(SERIALIZABLE_OBJECT_FILTER,
        "org.apache.geode.internal.cache.UnitTestValueHolder");
    return properties;
  }

  @Test
  public void testLargeGetAllFromServer() throws Throwable {
    final Host host = Host.getHost(0);
    final VM server = host.getVM(0);
    final VM client = host.getVM(1);
    final String regionName = getUniqueName();
    final int serverPort = AvailablePortHelper.getRandomAvailableTCPPort();
    final String serverHost = NetworkUtils.getServerHostName(server.getHost());

    createBridgeServer(server, regionName, serverPort, false, false);

    createBridgeClient(client, regionName, serverHost, new int[] {serverPort}, true);

    final int VALUE_SIZE = 1024 * 2/* *1024 */;

    final int VALUE_COUNT = 100;

    client.invoke(new CacheSerializableRunnable("put entries on server") {
      @Override
      public void run2() throws CacheException {
        final byte[] VALUE = new byte[VALUE_SIZE];
        for (int i = 0; i < VALUE_SIZE; i++) {
          VALUE[i] = (byte) i;
        }
        Region region = getRootRegion(regionName);
        for (int i = 0; i < VALUE_COUNT; i++) {
          region.put("k" + i, new UnitTestValueHolder(VALUE));
        }
      }
    });

    // force them to be deserialized on server
    server.invoke(new CacheSerializableRunnable("deserialize entries on server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion(regionName);
        for (int i = 0; i < VALUE_COUNT; i++) {
          region.get("k" + i);
        }
      }
    });

    CacheSerializableRunnable clientGetAll =
        new CacheSerializableRunnable("Get all entries from server") {
          @Override
          public void run2() throws CacheException {
            // Build collection of keys
            Collection keys = new ArrayList();
            for (int i = 0; i < VALUE_COUNT; i++) {
              keys.add("k" + i);
            }

            // Invoke getAll
            Region region = getRootRegion(regionName);
            final int GET_COUNT = 10/* 0 */;
            long start = System.currentTimeMillis();
            Map result = null;
            for (int i = 0; i < GET_COUNT; i++) {
              result = null; // allow gc to get rid of previous map before deserializing the next
                             // one
              result = region.getAll(keys);
            }
            long end = System.currentTimeMillis();
            long totalBytesRead = ((long) GET_COUNT * VALUE_COUNT * VALUE_SIZE);
            long elapsedMillis = (end - start);
            System.out.println("PERF: read " + totalBytesRead + " bytes in " + elapsedMillis
                + " millis. bps=" + (((double) totalBytesRead / elapsedMillis) * 1000));

            // Verify result size is correct
            assertEquals(VALUE_COUNT, result.size());

            final byte[] EXPECTED = new byte[VALUE_SIZE];
            for (int i = 0; i < VALUE_SIZE; i++) {
              EXPECTED[i] = (byte) i;
            }
            // Verify the result contains each key,
            // and the value for each key is correct
            // (the server has a loader that returns the key as the value)
            for (final Object o : keys) {
              String key = (String) o;
              assertTrue(result.containsKey(key));
              Object value = result.get(key);
              if (value instanceof UnitTestValueHolder) {
                Object v = ((UnitTestValueHolder) value).getValue();
                if (v instanceof byte[]) {
                  byte[] bytes = (byte[]) v;
                  if (bytes.length != VALUE_SIZE) {
                    fail("expected value for key " + key + " to be an array of size " + (VALUE_SIZE)
                        + " but it was: " + bytes.length);
                  }
                  if (!Arrays.equals(EXPECTED, bytes)) {
                    fail("expected bytes=" + Arrays.toString(bytes) + " to be expected="
                        + Arrays.toString(EXPECTED));
                  }
                } else {
                  fail("expected v for key " + key + " to be a byte array but it was: " + v);
                }
              } else {
                fail("expected value for key " + key + " to be a UnitTestValueHolder but it was: "
                    + value);
              }
            }
          }
        };
    // Run getAll
    {
      final int THREAD_COUNT = 4;
      AsyncInvocation<?>[] ais = new AsyncInvocation[THREAD_COUNT];
      for (int i = 0; i < THREAD_COUNT; i++) {
        ais[i] = client.invokeAsync(clientGetAll);
      }
      for (int i = 0; i < THREAD_COUNT; i++) {
        ais[i].getResult();
      }
    }
    stopBridgeServer(server);
  }

  @Test
  public void testGetAllWithCallbackFromServer() throws Exception {
    final Host host = Host.getHost(0);
    final VM server = host.getVM(0);
    final VM client = host.getVM(1);
    final String regionName = getUniqueName();
    final int serverPort = getRandomAvailableTCPPort();
    final String serverHost = NetworkUtils.getServerHostName(server.getHost());

    createBridgeServer(server, regionName, serverPort, false, true);

    createBridgeClient(client, regionName, serverHost, new int[] {serverPort});

    // Run getAll
    client.invoke(new CacheSerializableRunnable("Get all entries from server") {
      @Override
      public void run2() throws CacheException {
        // Build collection of keys
        Collection keys = new ArrayList();
        for (int i = 0; i < 5; i++) {
          keys.add("key-" + i);
        }

        keys.add(ClientServerTestCase.NON_EXISTENT_KEY); // this will not be load CacheLoader

        // Invoke getAll
        Region region = getRootRegion(regionName);
        Map result = region.getAll(keys, CALLBACK_ARG);

        // Verify result size is correct
        assertEquals(6, result.size());

        // Verify the result contains each key,
        // and the value for each key is correct
        // (the server has a loader that returns the key as the value)
        for (final Object o : keys) {
          String key = (String) o;
          assertTrue(result.containsKey(key));
          Object value = result.get(key);
          if (!key.equals(ClientServerTestCase.NON_EXISTENT_KEY)) {
            assertEquals(key, value);
          } else {
            assertEquals(null, value);
          }
        }

        assertEquals(null, region.get(ClientServerTestCase.NON_EXISTENT_KEY));
      }
    });

    stopBridgeServer(server);
  }

  @Test
  public void testGetSomeFromServer() throws Exception {
    testGetFromServer(2);
  }

  @Test
  public void testGetAllFromClient() throws Exception {
    testGetFromServer(5);
  }

  @Test
  public void testGetAllFromServerWithPR() throws Exception {
    final Host host = Host.getHost(0);
    final VM server1 = host.getVM(0);
    final VM server2 = host.getVM(1);
    final VM client = host.getVM(2);
    final String regionName = getUniqueName();
    int[] ports = AvailablePortHelper.getRandomAvailableTCPPorts(2);
    final int server1Port = ports[0];
    final int server2Port = ports[1];
    final String serverHost = NetworkUtils.getServerHostName(server1.getHost());

    createBridgeServer(server1, regionName, server1Port, true, false);

    createBridgeServer(server2, regionName, server2Port, true, false);

    createBridgeClient(client, regionName, serverHost, new int[] {server1Port, server2Port});

    // Run getAll
    client.invoke(new CacheSerializableRunnable("Get all entries from server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion(regionName);
        for (int i = 0; i < 200; i++) {
          region.put(i, i);
        }

        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
        // Build collection of keys
        Collection keys = new ArrayList();
        for (int i = 0; i < 5; i++) {
          keys.add("key-" + i);
        }
        keys.add(ClientServerTestCase.NON_EXISTENT_KEY); // this will not be load CacheLoader

        // Invoke getAll

        Map result = region.getAll(keys);


        // Verify result size is correct
        assertEquals(6, result.size());

        // Verify the result contains each key,
        // and the value for each key is correct
        // (the server has a loader that returns the key as the value)
        for (final Object o : keys) {
          String key = (String) o;
          assertTrue(result.containsKey(key));
          Object value = result.get(key);
          if (!key.equals(ClientServerTestCase.NON_EXISTENT_KEY)) {
            assertEquals(key, value);
          } else {
            assertEquals(null, value);
          }
        }
        assertEquals(null, region.get(ClientServerTestCase.NON_EXISTENT_KEY));
      }
    });

    stopBridgeServer(server1);

    stopBridgeServer(server2);
  }

  private void testGetFromServer(final int numLocalValues) {
    final Host host = Host.getHost(0);
    final VM server = host.getVM(0);
    final VM client = host.getVM(1);
    final String regionName = getUniqueName();
    final int serverPort = getRandomAvailableTCPPort();
    final String serverHost = NetworkUtils.getServerHostName(server.getHost());

    createBridgeServer(server, regionName, serverPort, false, false);

    createBridgeClient(client, regionName, serverHost, new int[] {serverPort});

    // Put some entries from the client
    client.invoke(new CacheSerializableRunnable("Put entries from client") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion(regionName);
        for (int i = 0; i < numLocalValues; i++) {
          region.put("key-" + i, "value-from-client-" + i);
        }
      }
    });

    // Run getAll
    client.invoke(new CacheSerializableRunnable("Get all entries from server") {
      @Override
      public void run2() throws CacheException {
        // Build collection of keys
        Collection keys = new ArrayList();
        for (int i = 0; i < 5; i++) {
          keys.add("key-" + i);
        }

        // Invoke getAll
        Region region = getRootRegion(regionName);
        Map result = region.getAll(keys);

        // Verify result size is correct
        assertEquals(5, result.size());

        // Verify the result contains each key,
        // and the value for each key is correct
        // (the server has a loader that returns the key as the value)
        // (the local value contains the phrase 'from client')
        int i = 0;
        for (Iterator it = keys.iterator(); it.hasNext(); i++) {
          String key = (String) it.next();
          assertTrue(result.containsKey(key));
          Object value = result.get(key);
          if (i < numLocalValues) {
            assertEquals("value-from-client-" + i, value);
          } else {
            assertEquals(key, value);
          }
        }
      }
    });

    // client may see "server unreachable" exceptions after this
    IgnoredException.addIgnoredException("Server unreachable", client);
    stopBridgeServer(server);
  }

  @Test
  public void testGetAllWithExtraKeyFromServer() throws Exception {
    final Host host = Host.getHost(0);
    final VM server = host.getVM(0);
    final VM client = host.getVM(1);
    final String regionName = getUniqueName();
    final int serverPort = getRandomAvailableTCPPort();
    final String serverHost = NetworkUtils.getServerHostName(server.getHost());
    final int numLocalValues = 101;

    createBridgeServerWithoutLoader(server, regionName, serverPort, false);

    createBridgeClient(client, regionName, serverHost, new int[] {serverPort});

    // Put some entries from the client
    client.invoke(new CacheSerializableRunnable("Put entries from client") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion(regionName);
        for (int i = 0; i < numLocalValues; i++) {
          region.put("key-" + i, "value-from-client-" + i);
        }
      }
    });

    server.invoke(new CacheSerializableRunnable("Put entries from server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion(regionName);
        for (int i = numLocalValues; i < numLocalValues * 2; i++) {
          region.put("key-" + i, "value-from-server-" + i);
        }
        region.getCache().getLogger().fine("The region entries in server " + region.entrySet());
      }
    });


    // Run getAll
    client.invoke(new CacheSerializableRunnable("Get all entries from server") {
      @Override
      public void run2() throws CacheException {
        // Build collection of keys
        Collection keys = new ArrayList();
        for (int i = 0; i < numLocalValues * 3; i++) {
          keys.add("key-" + i);
        }

        // Invoke getAll
        Region region = getRootRegion(regionName);
        region.getCache().getLogger()
            .fine("The region entries in client before getAll " + region.entrySet());
        assertEquals(region.entrySet().size(), numLocalValues);
        Map result = region.getAll(keys);
        assertEquals(region.entrySet().size(), 2 * numLocalValues);
        region.getCache().getLogger()
            .fine("The region entries in client after getAll " + region.entrySet());

        // Verify result size is correct
        assertEquals(3 * numLocalValues, result.size());

        // Verify the result contains each key,
        // and the value for each key is correct
        int i = 0;
        for (Iterator it = keys.iterator(); it.hasNext(); i++) {
          String key = (String) it.next();
          assertTrue(result.containsKey(key));
          Object value = result.get(key);
          if (i < numLocalValues) {
            assertEquals("value-from-client-" + i, value);
          } else if (i < 2 * numLocalValues) {
            assertEquals("value-from-server-" + i, value);
          } else {
            assertEquals(null, value);
          }
        }
      }
    });

    stopBridgeServer(server);
  }

  private void createBridgeServer(VM server, final String regionName, final int serverPort,
      final boolean createPR, final boolean expectCallback) {
    createBridgeServer(server, regionName, serverPort, createPR, expectCallback, false);
  }


  private void createBridgeServer(VM server, final String regionName, final int serverPort,
      final boolean createPR, final boolean expectCallback, final boolean offheap) {
    server.invoke(new CacheSerializableRunnable("Create server") {
      @Override
      public void run2() throws CacheException {
        // Create DS
        Properties config = getDistributedSystemProperties();
        if (offheap) {
          config.setProperty(OFF_HEAP_MEMORY_SIZE, "350m");
        }
        getSystem(config);

        // Create Region
        AttributesFactory factory = new AttributesFactory();
        if (offheap) {
          factory.setOffHeap(true);
        }
        if (expectCallback) {
          factory.setCacheLoader(new CallbackCacheServerCacheLoader());
        } else {
          factory.setCacheLoader(new CacheServerCacheLoader());
        }
        if (createPR) {
          factory.setDataPolicy(DataPolicy.PARTITION);
          factory.setPartitionAttributes((new PartitionAttributesFactory()).create());
        } else {
          factory.setScope(Scope.DISTRIBUTED_ACK);
          factory.setDataPolicy(DataPolicy.REPLICATE);
        }
        Region region = createRootRegion(regionName, factory.create());
        if (createPR) {
          assertTrue(region instanceof PartitionedRegion);
        }
        try {
          Cache cache = getCache();
          CacheServer bridge = cache.addCacheServer();
          bridge.setPort(serverPort);
          // for off-heap I want the server to use a selector
          bridge.setMaxThreads(offheap ? 16 : getMaxThreads());
          bridge.start();
        } catch (Exception e) {
          Assert.fail("While starting CacheServer", e);
        }
      }
    });
  }

  private static final String CALLBACK_ARG = "ClientServerGetAllDUnitTestCB";

  private static class CallbackCacheServerCacheLoader extends CacheServerCacheLoader {
    @Override
    public Object load2(LoaderHelper helper) {
      if (helper.getArgument() instanceof String) {
        if (!CALLBACK_ARG.equals(helper.getArgument())) {
          fail("Expected " + helper.getArgument() + " to be " + CALLBACK_ARG);
        }
      } else {
        if (!helper.getKey().equals(ClientServerTestCase.NON_EXISTENT_KEY)) {
          fail("Expected callback arg to be " + CALLBACK_ARG + " but it was null");
        }
      }
      return super.load2(helper);
    }
  }

  private void createBridgeServerWithoutLoader(VM server, final String regionName,
      final int serverPort, final boolean createPR) {
    server.invoke(new CacheSerializableRunnable("Create server") {
      @Override
      public void run2() throws CacheException {
        // Create DS
        Properties config = new Properties();
        config.setProperty(LOCATORS,
            "localhost[" + DistributedTestUtils.getDUnitLocatorPort() + "]");
        getSystem(config);

        // Create Region
        AttributesFactory factory = new AttributesFactory();
        if (createPR) {
          factory.setDataPolicy(DataPolicy.PARTITION);
          factory.setPartitionAttributes((new PartitionAttributesFactory()).create());
        } else {
          factory.setScope(Scope.DISTRIBUTED_ACK);
          factory.setDataPolicy(DataPolicy.REPLICATE);
        }
        Region region = createRootRegion(regionName, factory.create());
        if (createPR) {
          assertTrue(region instanceof PartitionedRegion);
        }
        try {
          startBridgeServer(serverPort);
          System.out.println("Started bridger server ");
        } catch (Exception e) {
          Assert.fail("While starting CacheServer", e);
        }
      }
    });
  }

  private void createBridgeClient(VM client, final String regionName, final String serverHost,
      final int[] serverPorts) {
    createBridgeClient(client, regionName, serverHost, serverPorts, false);
  }

  private void createBridgeClient(VM client, final String regionName, final String serverHost,
      final int[] serverPorts, final boolean proxy) {
    client.invoke(new CacheSerializableRunnable("Create client") {
      @Override
      public void run2() throws CacheException {
        // Create DS
        Properties config = getDistributedSystemProperties();
        ClientCache clientCache = getClientCache(new ClientCacheFactory(config));

        // Create Region
        ClientRegionFactory factory = clientCache.createClientRegionFactory(
            proxy ? ClientRegionShortcut.PROXY : ClientRegionShortcut.CACHING_PROXY);
        {
          PoolFactory pf = PoolManager.createFactory();
          for (final int serverPort : serverPorts) {
            pf.addServer(serverHost, serverPort);
          }
          if (proxy) {
            pf.setReadTimeout(30000);
          }
          pf.create("myPool");
        }
        factory.setPoolName("myPool");
        factory.create(regionName);
      }
    });
  }

  private void stopBridgeServer(VM server) {
    server.invoke(new CacheSerializableRunnable("Stop Server") {
      @Override
      public void run2() throws CacheException {
        stopBridgeServers(getCache());
      }
    });
  }

  private void checkServerForOrphans(VM server, final String regionName) {
    server.invoke(new CacheSerializableRunnable("Stop Server") {
      @Override
      public void run2() throws CacheException {
        Region region = getRootRegion(regionName);
        region.close();
        OffHeapTestUtil.checkOrphans(getCache());
      }
    });

  }
}
