/*
 * 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.tier.sockets;

import static org.apache.geode.cache.Region.SEPARATOR;
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.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

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

import org.apache.geode.GemFireConfigException;
import org.apache.geode.GemFireIOException;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.InterestResultPolicy;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionShortcut;
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.ClientRegionShortcut;
import org.apache.geode.cache.client.NoAvailableServersException;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.Connection;
import org.apache.geode.cache.client.internal.Op;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.client.internal.QueueConnectionImpl;
import org.apache.geode.cache.client.internal.RegisterInterestTracker;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
import org.apache.geode.logging.internal.OSProcess;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.dunit.DistributedTestUtils;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.dunit.internal.DUnitLauncher;
import org.apache.geode.test.junit.categories.ClientServerTest;
import org.apache.geode.test.version.VersionManager;
import org.apache.geode.test.version.VmConfiguration;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Tests client server corner cases between Region and Pool
 */
@Category({ClientServerTest.class})
public class ClientServerMiscDUnitTestBase extends JUnit4CacheTestCase {

  protected static PoolImpl pool = null;

  protected static Connection conn = null;

  static Cache static_cache;

  static int PORT1;

  private static final String k1 = "k1";

  private static final String k2 = "k2";

  static final String server_k1 = "server-k1";

  static final String server_k2 = "server-k2";

  static final String REGION_NAME1 = "ClientServerMiscDUnitTest_region1";

  static final String REGION_NAME2 = "ClientServerMiscDUnitTest_region2";

  private static final String PR_REGION_NAME = "ClientServerMiscDUnitTest_PRregion";

  private static Host host;

  protected static VM server1;

  protected static VM server2;

  private static RegionAttributes attrs;

  // variables for concurrent map API test
  Properties props = new Properties();
  private final int putRange_1Start = 1;
  private final int putRange_1End = 5;
  private final int putRange_2Start = 6;
  private final int putRange_2End = 10;


  // configuration for client caches for backward-compatibility testing
  final VmConfiguration clientVmConfiguration;

  public ClientServerMiscDUnitTestBase() {
    this(VmConfiguration.current());
  }

  public ClientServerMiscDUnitTestBase(VmConfiguration clientVmConfiguration) {
    this.clientVmConfiguration = clientVmConfiguration;
  }

  @Override
  public final void postSetUp() {
    host = Host.getHost(0);
    server1 = host.getVM(VersionManager.CURRENT_VERSION, 2);
    server2 = host.getVM(VersionManager.CURRENT_VERSION, 3);
  }

  int initServerCache(boolean notifyBySub) {
    return initServerCache(notifyBySub, false);
  }

  int initServerCache2() {
    return initServerCache2(false);
  }

  private int initServerCache(boolean notifyBySub, boolean isHA) {
    return initServerCache(notifyBySub, server1, isHA);
  }

  private int initServerCache2(boolean isHA) {
    return initServerCache(true, server2, isHA);
  }

  int initServerCache(boolean notifyBySub, VM vm, boolean isHA) {
    int port = getRandomAvailableTCPPort();
    return vm.invoke(() -> createServerCache(notifyBySub, getMaxThreads(), isHA, port));
  }

  @Test
  public void testConcurrentOperationsWithDRandPR() {
    int port1 = initServerCache(true); // vm0
    int port2 = initServerCache2(); // vm1
    String serverName = NetworkUtils.getServerHostName();
    host.getVM(clientVmConfiguration, 0).invoke(() -> createClientCacheV(serverName, port1));
    host.getVM(clientVmConfiguration, 1).invoke(() -> createClientCacheV(serverName, port2));
    LogService.getLogger()
        .info("Testing concurrent map operations from a client with a distributed region");
    concurrentMapTest(host.getVM(clientVmConfiguration, 0), SEPARATOR + REGION_NAME1);
    // TODO add verification in vm1
    LogService.getLogger()
        .info("Testing concurrent map operations from a client with a partitioned region");
    concurrentMapTest(host.getVM(clientVmConfiguration, 0), SEPARATOR + PR_REGION_NAME);
    // TODO add verification in vm1
  }

  /**
   * When a client's subscription thread connects to a server it should receive the server's
   * pingInterval setting. This is used by the client to set a read-timeout in order to avoid
   * hanging should the server's machine crash.
   */
  @Test
  public void testClientReceivesPingIntervalSetting() {
    VM clientVM = Host.getHost(0).getVM(clientVmConfiguration, 0);

    final int port = initServerCache(true);
    final String host = NetworkUtils.getServerHostName();

    clientVM.invoke("create client cache and verify", () -> {
      createClientCacheAndVerifyPingIntervalIsSet(host, port);
    });
  }

  void createClientCacheAndVerifyPingIntervalIsSet(String host, int port) throws Exception {
    PoolImpl pool = null;
    try {
      Properties props = new Properties();
      props.setProperty(MCAST_PORT, "0");
      props.setProperty(LOCATORS, "");

      createCache(props);

      pool = (PoolImpl) PoolManager.createFactory().addServer(host, port)
          .setSubscriptionEnabled(true).setReadTimeout(1000)
          .setSocketBufferSize(32768).setMinConnections(1).setSubscriptionRedundancy(-1)
          .setPingInterval(2000).create("test pool");

      Region<Object, Object> region = cache.createRegionFactory(RegionShortcut.LOCAL)
          .setPoolName("test pool").create(REGION_NAME1);
      region.registerInterest(".*");

      /** get the subscription connection and verify that it has the correct timeout setting */
      QueueConnectionImpl primaryConnection = (QueueConnectionImpl) pool.getPrimaryConnection();
      int pingInterval = ((CacheClientUpdater) primaryConnection.getUpdater())
          .getServerQueueStatus().getPingInterval();
      assertNotEquals(0, pingInterval);
      assertEquals(CacheClientNotifier.getClientPingInterval(), pingInterval);
    } finally {
      cache.close();
    }

  }

  @Test
  public void testConcurrentOperationsWithDRandPRandEmptyClient() {
    int port1 = initServerCache(true); // vm0
    int port2 = initServerCache2(); // vm1
    String serverName = NetworkUtils.getServerHostName();
    host.getVM(clientVmConfiguration, 0).invoke(() -> createEmptyClientCache(serverName, port1));
    host.getVM(clientVmConfiguration, 1).invoke(() -> createClientCacheV(serverName, port2));
    LogService.getLogger()
        .info("Testing concurrent map operations from a client with a distributed region");
    concurrentMapTest(host.getVM(clientVmConfiguration, 0), SEPARATOR + REGION_NAME1);
    // TODO add verification in vm1
    LogService.getLogger()
        .info("Testing concurrent map operations from a client with a partitioned region");
    concurrentMapTest(host.getVM(clientVmConfiguration, 0), SEPARATOR + PR_REGION_NAME);
    // TODO add verification in vm1
  }

  /**
   * Do putIfAbsent(), replace(Object, Object), replace(Object, Object, Object), remove(Object,
   * Object) operations
   */
  private void concurrentMapTest(final VM clientVM, final String rName) {

    // String exceptionStr = "";
    clientVM.invoke(new CacheSerializableRunnable("doConcurrentMapOperations") {
      @Override
      public void run2() throws CacheException {
        Cache cache = getCache();
        final Region pr = cache.getRegion(rName);
        assertNotNull(rName + " not created", pr);
        boolean isEmpty = pr.getAttributes().getDataPolicy() == DataPolicy.EMPTY;

        // test successful putIfAbsent
        for (int i = putRange_1Start; i <= putRange_1End; i++) {
          Object putResult = pr.putIfAbsent(Integer.toString(i), Integer.toString(i));
          assertNull("Expected null, but got " + putResult + " for key " + i, putResult);
        }
        int size;
        if (!isEmpty) {
          size = pr.size();
          assertEquals("Size doesn't return expected value", putRange_1End, size);
          assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty());
        }

        // test unsuccessful putIfAbsent
        for (int i = putRange_1Start; i <= putRange_1End; i++) {
          Object putResult = pr.putIfAbsent(Integer.toString(i), Integer.toString(i + 1));
          assertEquals("for i=" + i, Integer.toString(i), putResult);
          assertEquals("for i=" + i, Integer.toString(i), pr.get(Integer.toString(i)));
        }
        if (!isEmpty) {
          size = pr.size();
          assertEquals("Size doesn't return expected value", putRange_1End, size);
          assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty());
        }

        // test successful replace(key, oldValue, newValue)
        for (int i = putRange_1Start; i <= putRange_1End; i++) {
          boolean replaceSucceeded =
              pr.replace(Integer.toString(i), Integer.toString(i), "replaced" + i);
          assertTrue("for i=" + i, replaceSucceeded);
          assertEquals("for i=" + i, "replaced" + i, pr.get(Integer.toString(i)));
        }
        if (!isEmpty) {
          size = pr.size();
          assertEquals("Size doesn't return expected value", putRange_1End, size);
          assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty());
        }

        // test unsuccessful replace(key, oldValue, newValue)
        for (int i = putRange_1Start; i <= putRange_2End; i++) {
          boolean replaceSucceeded = pr.replace(Integer.toString(i), Integer.toString(i), // wrong
                                                                                          // expected
                                                                                          // old
                                                                                          // value
              "not" + i);
          assertFalse("for i=" + i, replaceSucceeded);
          assertEquals("for i=" + i, i <= putRange_1End ? "replaced" + i : null,
              pr.get(Integer.toString(i)));
        }
        if (!isEmpty) {
          size = pr.size();
          assertEquals("Size doesn't return expected value", putRange_1End, size);
          assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty());
        }

        // test successful replace(key, value)
        for (int i = putRange_1Start; i <= putRange_1End; i++) {
          Object replaceResult = pr.replace(Integer.toString(i), "twice replaced" + i);
          assertEquals("for i=" + i, "replaced" + i, replaceResult);
          assertEquals("for i=" + i, "twice replaced" + i, pr.get(Integer.toString(i)));
        }
        if (!isEmpty) {
          size = pr.size();
          assertEquals("Size doesn't return expected value", putRange_1End, size);
          assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty());
        }

        // test unsuccessful replace(key, value)
        for (int i = putRange_2Start; i <= putRange_2End; i++) {
          Object replaceResult = pr.replace(Integer.toString(i), "thrice replaced" + i);
          assertNull("for i=" + i, replaceResult);
          assertNull("for i=" + i, pr.get(Integer.toString(i)));
        }
        if (!isEmpty) {
          size = pr.size();
          assertEquals("Size doesn't return expected value", putRange_1End, size);
          assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty());
        }

        // test unsuccessful remove(key, value)
        for (int i = putRange_1Start; i <= putRange_2End; i++) {
          boolean removeResult = pr.remove(Integer.toString(i), Integer.toString(-i));
          assertFalse("for i=" + i, removeResult);
          assertEquals("for i=" + i, i <= putRange_1End ? "twice replaced" + i : null,
              pr.get(Integer.toString(i)));
        }
        if (!isEmpty) {
          size = pr.size();
          assertEquals("Size doesn't return expected value", putRange_1End, size);
          assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty());
        }

        // test successful remove(key, value)
        for (int i = putRange_1Start; i <= putRange_1End; i++) {
          boolean removeResult = pr.remove(Integer.toString(i), "twice replaced" + i);
          assertTrue("for i=" + i, removeResult);
          assertNull("for i=" + i, pr.get(Integer.toString(i)));
        }
        if (!isEmpty) {
          size = pr.size();
          assertEquals("Size doesn't return expected value", 0, size);
          pr.localClear();
          assertTrue("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty());
        }

        if (!isEmpty) {
          // bug #42169 - entry not updated on server when locally destroyed on client
          String key42169 = "key42169";
          pr.put(key42169, "initialValue42169");
          pr.localDestroy(key42169);
          boolean success = pr.replace(key42169, "initialValue42169", "newValue42169");
          assertTrue("expected replace to succeed", success);
          pr.destroy(key42169);
          pr.put(key42169, "secondRound");
          pr.localDestroy(key42169);
          Object result = pr.putIfAbsent(key42169, null);
          assertEquals("expected putIfAbsent to fail", result, "secondRound");
          pr.destroy(key42169);
        }

        if (isEmpty) {
          String key41265 = "key41265";
          boolean success = pr.remove(key41265, null);
          assertFalse("expected remove to fail because key does not exist", success);
        }

        // test null values

        // putIfAbsent with null value creates invalid entry
        Object oldValue = pr.putIfAbsent("keyForNull", null);
        assertNull(oldValue);
        if (!isEmpty) {
          assertTrue(pr.containsKey("keyForNull"));
          assertTrue(!pr.containsValueForKey("keyForNull"));
        }

        // replace allows null value for oldValue, meaning invalidated entry
        assertTrue(pr.replace("keyForNull", null, "no longer invalid"));

        // replace does not allow null value for new value
        assertThatThrownBy(() -> pr.replace("keyForNull", "no longer invalid", null))
            .isExactlyInstanceOf(NullPointerException.class);

        // other variant of replace does not allow null value for new value
        assertThatThrownBy(() -> pr.replace("keyForNull", null))
            .isExactlyInstanceOf(NullPointerException.class);

        // replace with null oldvalue matches invalidated entry
        pr.putIfAbsent("otherKeyForNull", null);
        CachePerfStats stats = ((GemFireCacheImpl) pr.getCache()).getCachePerfStats();

        Number puts = getNumPuts(stats);
        boolean success = pr.replace("otherKeyForNull", null, "no longer invalid");
        assertTrue(success);

        Number newputs = getNumPuts(stats);
        assertEquals("stats not updated properly or replace malfunctioned", newputs.longValue(),
            puts.longValue() + 1);

      }

      public Number getNumPuts(CachePerfStats stats) {
        Method getPutsMethod = null;
        try {
          getPutsMethod = stats.getClass().getMethod("getPuts");
        } catch (NoSuchMethodException e) {
          fail(e.getMessage());
        }

        Number puts = null;
        try {
          puts = (Number) getPutsMethod.invoke(stats);
        } catch (IllegalAccessException e) {
          fail(e.getMessage());
        } catch (InvocationTargetException e) {
          fail(e.getMessage());
        }

        return puts;
      }
    });
  }

  /**
   * Test two regions: notify by subscription is true. For region1 the interest list is empty , for
   * region 2 the intetest list is all keys. If an update/create is made on region1 , the client
   * should not receive any. If the create/update is on region2 , the client should receive the
   * update.
   */
  @Test
  public void testForTwoRegionHavingDifferentInterestList() {
    // start server first
    PORT1 = initServerCache(true);
    int serverPort = PORT1;
    VM client1 = Host.getHost(0).getVM(clientVmConfiguration, 1);
    String hostname = NetworkUtils.getServerHostName();
    client1.invoke("create client1 cache", () -> {
      createClientCache(hostname, serverPort);
      populateCache();
      registerInterest();
    });

    server1.invoke("putting entries in server1", ClientServerMiscDUnitTestBase::put);

    client1.invoke(ClientServerMiscDUnitTestBase::verifyUpdates);
  }

  /**
   * Test two regions: notify by subscription is true. Both the regions have registered interest in
   * all the keys. Now close region1 on the client. The region1 should get removed from the interest
   * list on CCP at server. Any update on region1 on server should not get pushed to the client.
   * Ensure that the message related is not added to the client's queue at all ( which is diferent
   * from not receiving a callbak on the client). If an update on region2 is made on the server ,
   * then client should receive the calback
   */
  @Test
  public void testForTwoRegionHavingALLKEYSInterest() throws Exception {
    // start server first
    PORT1 = initServerCache(true);
    createClientCache(NetworkUtils.getServerHostName(), PORT1);
    populateCache();
    registerInterestInBothTheRegions();
    closeRegion1();
    Wait.pause(6000);
    server1.invoke(ClientServerMiscDUnitTestBase::verifyInterestListOnServer);
    server1.invoke(ClientServerMiscDUnitTestBase::put);
    // pause(5000);
    verifyUpdatesOnRegion2();
  }

  /**
   * Test two regions: notify by subscription is true. Both the regions have registered interest in
   * all the keys. Close both the regions. When the last region is closed , it should close the
   * ConnectionProxy on the client , close all the server connection threads on the server & remove
   * the CacheClientProxy from the CacheClient notifier
   */
  @Test
  public void testRegionClose() throws Exception {
    // start server first
    PORT1 = initServerCache(true);
    pool = (PoolImpl) createClientCache(NetworkUtils.getServerHostName(), PORT1);
    populateCache();
    registerInterestInBothTheRegions();
    closeBothRegions();

    assertFalse(pool.isDestroyed());
    pool.destroy();
    assertTrue(pool.isDestroyed());
    server1.invoke(ClientServerMiscDUnitTestBase::verifyNoCacheClientProxyOnServer);

  }

  /**
   * Test two regions: notify by subscription is true. Both the regions have registered interest in
   * all the keys. Destroy region1 on the client. It should reach the server , kill the region on
   * the server , propagate it to the interested clients , but it should keep CacheClient Proxy
   * alive. Destroy Region2 . It should reach server , close conenction proxy , destroy the region2
   * on the server , remove the cache client proxy from the cache client notifier & propagate it to
   * the clients. Then create third region and verify that no CacheClientProxy is created on server
   */
  @Test
  public void testCCPDestroyOnLastDestroyRegion() throws Exception {
    PORT1 = initServerCache(true);
    PoolImpl pool =
        (PoolImpl) createClientCache(NetworkUtils.getServerHostName(), PORT1);
    destroyRegion1();
    // pause(5000);
    server1.invoke(
        () -> ClientServerMiscDUnitTestBase.verifyCacheClientProxyOnServer(REGION_NAME1));
    Connection conn = pool.acquireConnection();
    assertNotNull(conn);
    assertEquals(1, pool.getConnectedServerCount());
    assertFalse(pool.isDestroyed());
    destroyRegion2();
    assertFalse(pool.isDestroyed());
    destroyPRRegion();
    assertFalse(pool.isDestroyed());
    pool.destroy();
    assertTrue(pool.isDestroyed());
    // pause(5000);
    server1.invoke(ClientServerMiscDUnitTestBase::verifyNoCacheClientProxyOnServer);

    assertThatThrownBy(() -> getCache().createRegion(REGION_NAME2, attrs))
        .isExactlyInstanceOf(IllegalStateException.class);
  }

  /**
   * Test two regions:If notify by subscription is false , both the regions should receive
   * invalidates for the updates on server in their respective regions
   *
   */
  @Test
  public void testInvalidatesPropagateOnTwoRegions() throws Exception {
    // start server first
    PORT1 = initServerCache(false);
    createClientCache(NetworkUtils.getServerHostName(), PORT1);
    registerInterestForInvalidatesInBothTheRegions();
    populateCache();
    server1.invoke(ClientServerMiscDUnitTestBase::put);
    // pause(5000);
    verifyInvalidatesOnBothRegions();

  }

  /**
   * Test for bug 43407, where LRU in the client caused an entry to be evicted with DESTROY(), then
   * the client invalidated the entry and did a get(). After the get() the entry was not seen to be
   * in the client's cache. This turned out to be expected behavior, but we now have this test to
   * guarantee that the product behaves as expected.
   */
  @Test
  public void testGetInClientCreatesEntry() throws Exception {
    // start server first
    PORT1 = initServerCache(false);
    createClientCache(NetworkUtils.getServerHostName(), PORT1);
    registerInterestForInvalidatesInBothTheRegions();
    Region region = static_cache.getRegion(REGION_NAME1);
    populateCache();
    region.put("invalidationKey", "invalidationValue");

    region.localDestroy("invalidationKey");
    assertThat(region.containsKey("invalidationKey")).isFalse();

    region.invalidate("invalidationKey");
    assertThat(region.containsKey("invalidationKey")).isTrue();

    Object value = region.get("invalidationKey");
    assertThat(value).isNull();
    assertThat(region.containsKeyOnServer("invalidationKey")).isTrue();
  }

  /**
   * GEODE-478 - large payloads are rejected by client->server
   */
  @Test
  public void testLargeMessageIsRejected() throws Exception {
    PORT1 = initServerCache(false);
    createClientCache(NetworkUtils.getServerHostName(), PORT1);
    Region region = static_cache.getRegion(REGION_NAME1);
    Op operation = cnx -> {
      throw new MessageTooLargeException("message is too big");
    };
    try {
      ((LocalRegion) region).getServerProxy().getPool().execute(operation);
    } catch (GemFireIOException e) {
      assertTrue(e.getCause() instanceof MessageTooLargeException);
      return;
    }
    fail("expected an exception to be thrown");
  }

  /**
   * Create cache, create pool, notify-by-subscription=false, create a region and on client and on
   * server. Do not attach pool to region , populate some entries on region both on client and
   * server. Update the entries on server the client. The client should not have entry invalidate.
   */
  @Test
  public void testInvalidatesPropagateOnRegionHavingNoPool() {
    // start server first
    PORT1 = initServerCache(false);
    Properties props = new Properties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(LOCATORS, "");
    new ClientServerMiscDUnitTestBase().createCache(props);
    String host = NetworkUtils.getServerHostName();
    PoolImpl p =
        (PoolImpl) PoolManager.createFactory().addServer(host, PORT1).setSubscriptionEnabled(true)
            .setReadTimeout(1000).setSocketBufferSize(32768)
            .setMinConnections(3).setSubscriptionRedundancy(-1).setPingInterval(2000)
            // .setRetryAttempts(5)
            // .setRetryInterval(2000)
            .create("testInvalidatesPropagateOnRegionHavingNoPool");

    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    // factory.setPoolName(p.getName());

    attrs = factory.create();
    final Region region1 = getCache().createRegion(REGION_NAME1, attrs);
    final Region region2 = getCache().createRegion(REGION_NAME2, attrs);
    assertNotNull(region1);
    assertNotNull(region2);
    pool = p;
    conn = pool.acquireConnection();
    assertNotNull(conn);

    populateCache();
    server1.invoke(ClientServerMiscDUnitTestBase::put);

    await().until(() -> {
      Object val = region1.getEntry(k1).getValue();
      return k1.equals(val);
    });

    await().until(() -> {
      Object val = region1.getEntry(k2).getValue();
      return k2.equals(val);
    });

    await().until(() -> {
      Object val = region2.getEntry(k1).getValue();
      return k1.equals(val);
    });

    await().until(() -> {
      Object val = region2.getEntry(k2).getValue();
      return k2.equals(val);
    });

    // assertIndexDetailsEquals(region2.getEntry(k2).getValue(), k2);
  }

  /**
   * Create proxy before cache creation, create cache, create two regions, attach same bridge writer
   * to both of the regions Region interests AL_KEYS on both the regions,
   * notify-by-subscription=true . The CCP should have both the regions in interest list.
   *
   */

  @Test
  public void testProxyCreationBeforeCacheCreation() {
    Properties props = new Properties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(LOCATORS, "");
    DistributedSystem ds = getSystem(props);
    assertNotNull(ds);
    ds.disconnect();
    ds = getSystem(props);
    PORT1 = initServerCache(true);
    String host = NetworkUtils.getServerHostName();
    Pool p = PoolManager.createFactory().addServer(host, PORT1).setSubscriptionEnabled(true)
        .setSubscriptionRedundancy(-1)
        // .setRetryAttempts(5)
        .create("testProxyCreationBeforeCacheCreationPool");

    Cache cache = getCache();
    assertNotNull(cache);

    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setPoolName(p.getName());

    RegionAttributes myAttrs = factory.create();
    Region region1 = cache.createRegion(REGION_NAME1, myAttrs);
    Region region2 = cache.createRegion(REGION_NAME2, myAttrs);
    assertNotNull(region1);
    assertNotNull(region2);
    // region1.registerInterest(CacheClientProxy.ALL_KEYS);
    region2.registerInterest("ALL_KEYS");
    Wait.pause(6000);
    server1.invoke(ClientServerMiscDUnitTestBase::verifyInterestListOnServer);

  }

  /**
   *
   * bug 35380: Cycling a DistributedSystem with an initialized pool causes interest registration
   * NPE
   *
   * Test Scenario:
   *
   * Create a DistributedSystem (DS1). Create a pool, initialize (creates a proxy with DS1 memberid)
   * Disconnect DS1. Create a DistributedSystem (DS2). Create a Region with pool, it attempts to
   * register interest using DS2 memberid, gets NPE.
   *
   */
  @Test
  public void testSystemCanBeCycledWithAnInitializedPool() {
    // work around GEODE-477
    addIgnoredException("Connection reset");
    Properties props = new Properties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(LOCATORS, "");
    DistributedSystem ds = getSystem(props);
    assertNotNull(ds);

    PORT1 = initServerCache(true);
    String host = NetworkUtils.getServerHostName();
    Pool p = PoolManager.createFactory().addServer(host, PORT1).setSubscriptionEnabled(true)
        .setSubscriptionRedundancy(-1)
        // .setRetryAttempts(5)
        .create("testBug35380Pool");

    Cache cache = getCache();

    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setPoolName(p.getName());

    RegionAttributes myAttrs = factory.create();
    Region region1 = cache.createRegion(REGION_NAME1, myAttrs);
    Region region2 = cache.createRegion(REGION_NAME2, myAttrs);
    assertNotNull(region1);
    assertNotNull(region2);

    region2.registerInterest("ALL_KEYS");

    ds.disconnect();
    Properties prop = new Properties();
    prop.setProperty(MCAST_PORT, "0");
    prop.setProperty(LOCATORS, "");
    ds = getSystem(prop);

    final Cache cacheForLamda = getCache();
    assertNotNull(cacheForLamda);

    AttributesFactory factory1 = new AttributesFactory();
    factory1.setScope(Scope.DISTRIBUTED_ACK);
    // reuse writer from prev DS
    factory1.setPoolName(p.getName());

    final RegionAttributes attrs1 = factory1.create();
    assertThatThrownBy(() -> cacheForLamda.createRegion(REGION_NAME1, attrs1))
        .isInstanceOfAny(IllegalStateException.class, DistributedSystemDisconnectedException.class);
  }

  @Test(expected = GemFireConfigException.class)
  public void clientIsPreventedFromConnectingToLocatorAsServer() {
    addIgnoredException("Improperly configured client detected", VM.getLocator());
    ClientCacheFactory clientCacheFactory = new ClientCacheFactory();
    clientCacheFactory.addPoolServer("localhost", DistributedTestUtils.getDUnitLocatorPort());
    clientCacheFactory.setPoolSubscriptionEnabled(true);
    getClientCache(clientCacheFactory);
    Region region = ((ClientCache) cache).createClientRegionFactory(ClientRegionShortcut.PROXY)
        .create(REGION_NAME1);
    region.registerInterest(k1);
  }


  private void createCache(Properties props) {
    createCacheV(props);
  }

  private Cache createCacheV(Properties props) {
