/*
 * 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.DELTA_PROPAGATION;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Properties;

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

import org.apache.geode.DeltaTestImpl;
import org.apache.geode.LogWriter;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.InterestPolicy;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.SubscriptionAttributes;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqAttributesFactory;
import org.apache.geode.cache.query.CqEvent;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.util.CqListenerAdapter;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.TestObjectWithIdentifier;
import org.apache.geode.test.awaitility.GeodeAwaitility;
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.WaitCriterion;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;
import org.apache.geode.test.junit.categories.SerializationTest;

/**
 * Test client to server flow for delta propogation
 *
 * @since GemFire 6.1
 */
@Category({ClientSubscriptionTest.class, SerializationTest.class})
public class ClientToServerDeltaDUnitTest extends JUnit4DistributedTestCase {
  /*
   * Test configuration one server one client also 2 server 2 client
   */
  private static Cache cache = null;

  private static LogWriter logger = null;

  VM server = null;
  VM server2 = null;
  VM client = null;
  VM client2 = null;

  private static final String KEY1 = "DELTA_KEY_1";

  private static final String REGION_NAME = "ClientToServerDeltaDunitTest_region";

  private static final int NO_PUT_OPERATION = 3;

  private static PoolImpl pool;

  private static Object[] putDelta = {"Anil", DeltaTestImpl.ERRONEOUS_STRING_FOR_FROM_DELTA, 100,
      DeltaTestImpl.ERRONEOUS_INT_FOR_TO_DELTA};

  private static int updates = 0;

  private static int cqUpdates = 0;

  private static int create = 0;

  private static Object firstUpdate = null;

  private static Object secondUpdate = null;

  private static Boolean error;

  private static boolean lastKeyReceived = false;

  private static Region region = null;

  private static CSDeltaTestImpl csDelta = null;

  public static String DELTA_KEY = "DELTA_KEY";

  private static final String[] CQs = new String[] {"select * from " + SEPARATOR + REGION_NAME,
      "select * from " + SEPARATOR + REGION_NAME + " where intVar = 0",
      "select * from " + SEPARATOR + REGION_NAME + " where intVar > 0",
      "select * from " + SEPARATOR + REGION_NAME + " where intVar < 0"};

  public static String LAST_KEY = "LAST_KEY";


  @Rule
  public DistributedRestoreSystemProperties restoreSystemProperties =
      new DistributedRestoreSystemProperties();

  @Override
  public final void postSetUp() throws Exception {
    disconnectAllFromDS();
    final Host host = Host.getHost(0);
    server = host.getVM(0);
    server2 = host.getVM(1);
    client = host.getVM(2);
    client2 = host.getVM(3);
  }

  @Override
  public final void preTearDown() throws Exception {
    // reset all flags
    DeltaTestImpl.resetDeltaInvokationCounters();
    server.invoke(() -> DeltaTestImpl.resetDeltaInvokationCounters());
    server2.invoke(() -> DeltaTestImpl.resetDeltaInvokationCounters());
    client.invoke(() -> DeltaTestImpl.resetDeltaInvokationCounters());
    client2.invoke(() -> DeltaTestImpl.resetDeltaInvokationCounters());
    // close the clients first
    client.invoke(() -> ClientToServerDeltaDUnitTest.closeCache());
    client2.invoke(() -> ClientToServerDeltaDUnitTest.closeCache());
    // then close the servers
    server.invoke(() -> ClientToServerDeltaDUnitTest.closeCache());
    server2.invoke(() -> ClientToServerDeltaDUnitTest.closeCache());
  }

  public void initialise(Boolean cq) {
    initialise(Boolean.TRUE/* clone */, null, cq, Boolean.TRUE/* RI */,
        Boolean.TRUE/* enable delta */);
  }

  public void initialise(Boolean clone, String[] queries, Boolean cq, Boolean RI,
      Boolean enableDelta) {
    Integer PORT1 = ((Integer) server.invoke(() -> ClientToServerDeltaDUnitTest
        .createServerCache(Boolean.TRUE, Boolean.FALSE, clone, enableDelta))).intValue();

    Integer PORT2 = ((Integer) server2.invoke(() -> ClientToServerDeltaDUnitTest
        .createServerCache(Boolean.TRUE, Boolean.FALSE, clone, enableDelta))).intValue();

    client.invoke(() -> ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server.getHost()), new Integer(PORT1), Boolean.FALSE,
        Boolean.FALSE, Boolean.FALSE));

    client2.invoke(() -> ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server2.getHost()), new Integer(PORT2), Boolean.TRUE,
        Boolean.FALSE, cq, queries, RI));
  }

  // Same as initialise() except listener flag is false.
  public void initialise2(Boolean clone, String[] queries, Boolean cq, Boolean RI,
      Boolean enableDelta) {
    Integer PORT1 = ((Integer) server.invoke(() -> ClientToServerDeltaDUnitTest
        .createServerCache(Boolean.FALSE, Boolean.FALSE, clone, enableDelta))).intValue();

    Integer PORT2 = ((Integer) server2.invoke(() -> ClientToServerDeltaDUnitTest
        .createServerCache(Boolean.FALSE, Boolean.FALSE, clone, enableDelta))).intValue();

    client.invoke(() -> ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server.getHost()), new Integer(PORT1), Boolean.FALSE,
        Boolean.FALSE, Boolean.FALSE));

    client2.invoke(() -> ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server2.getHost()), new Integer(PORT2), Boolean.TRUE,
        Boolean.FALSE, cq, queries, RI));
  }

  /**
   * This test does the following for single key (<b>failure of fromDelta- resending of full
   * object</b>):<br>
   * 1)Verifies that we donot loose attributes updates when delta fails<br>
   */
  @Test
  public void testSendingofFullDeltaObjectsWhenFromDeltaFails() {
    initialise(false);
    // set expected value on server
    server.invoke(() -> ClientToServerDeltaDUnitTest.setFirstSecondUpdate(putDelta[1],
        (Integer) putDelta[2]));
    client.invoke(() -> ClientToServerDeltaDUnitTest.putWithFromDeltaERR(KEY1));

    assertTrue("to Delta Propagation feature NOT used.",
        ((Boolean) client.invoke(() -> DeltaTestImpl.toDeltaFeatureUsed())).booleanValue());
    assertTrue("from Delta Propagation feature NOT used.",
        ((Boolean) server.invoke(() -> DeltaTestImpl.fromDeltaFeatureUsed())).booleanValue());

    assertFalse("Delta Propagation toDeltaFailed",
        ((Boolean) client.invoke(() -> DeltaTestImpl.isToDeltaFailure())).booleanValue());
    assertTrue("Delta Propagation fromDelta not Failed",
        ((Boolean) server.invoke(() -> DeltaTestImpl.isFromDeltaFailure())).booleanValue());

    boolean err =
        ((Boolean) server.invoke(() -> ClientToServerDeltaDUnitTest.getError())).booleanValue();
    assertFalse("validation fails", err);
  }

  /**
   * This test does the following for single key</br>
   * 1)Verifies that we donot loose attributes updates when delta fails<br>
   */
  @Test
  public void testPutForDeltaObjects() {
    initialise(false);
    // set expected value on server
    server.invoke(() -> ClientToServerDeltaDUnitTest.setFirstSecondUpdate(putDelta[0],
        (Integer) putDelta[2]));

    client.invoke(() -> ClientToServerDeltaDUnitTest.put(KEY1));

    assertTrue("to Delta Propagation feature NOT used.",
        ((Boolean) client.invoke(() -> DeltaTestImpl.toDeltaFeatureUsed())).booleanValue());
    assertTrue("from Delta Propagation feature NOT used.",
        ((Boolean) server.invoke(() -> DeltaTestImpl.fromDeltaFeatureUsed())).booleanValue());
    assertFalse("Delta Propagation toDeltaFailed",
        ((Boolean) client.invoke(() -> DeltaTestImpl.isToDeltaFailure())).booleanValue());
    assertFalse("Delta Propagation fromDeltaFailed",
        ((Boolean) server.invoke(() -> DeltaTestImpl.isFromDeltaFailure())).booleanValue());
    boolean err =
        ((Boolean) server.invoke(() -> ClientToServerDeltaDUnitTest.getError())).booleanValue();
    assertFalse("validation fails", err);
  }

  /**
   * This test does the following for single key (<b> full cycle</b>):<br>
   * 1)Verifies that client-server, peer-peer and server-client processing delta<br>
   */
  @Test
  public void testClientToClientDeltaPropagation() throws Exception {
    initialise(false);
    // set expected value on s1,s1 and c2
    server.invoke(() -> ClientToServerDeltaDUnitTest.setFirstSecondUpdate(putDelta[0],
        (Integer) putDelta[2]));
    server2.invoke(() -> ClientToServerDeltaDUnitTest.setFirstSecondUpdate(putDelta[0],
        (Integer) putDelta[2]));
    client2.invoke(() -> ClientToServerDeltaDUnitTest.setFirstSecondUpdate(putDelta[0],
        (Integer) putDelta[2]));

    client.invoke(() -> ClientToServerDeltaDUnitTest.putDelta(KEY1));

    Thread.sleep(5000);

    assertTrue("To Delta Propagation feature NOT used.",
        ((Boolean) client.invoke(() -> DeltaTestImpl.toDeltaFeatureUsed())).booleanValue());
    assertTrue("From Delta Propagation feature NOT used.",
        ((Boolean) server.invoke(() -> DeltaTestImpl.fromDeltaFeatureUsed())).booleanValue());
    // toDelta() should not be invoked
    assertFalse("To Delta Propagation feature used.",
        ((Boolean) server.invoke(() -> DeltaTestImpl.toDeltaFeatureUsed())).booleanValue());
    assertTrue("From Delta Propagation feature NOT used.",
        ((Boolean) server2.invoke(() -> DeltaTestImpl.fromDeltaFeatureUsed())).booleanValue());
    // toDelta() should not be invoked
    assertFalse("To Delta Propagation feature used.",
        ((Boolean) server2.invoke(() -> DeltaTestImpl.toDeltaFeatureUsed())).booleanValue());
    assertTrue("from Delta Propagation feature NOT used.",
        ((Boolean) client2.invoke(() -> DeltaTestImpl.fromDeltaFeatureUsed())).booleanValue());

    boolean err =
        ((Boolean) server.invoke(() -> ClientToServerDeltaDUnitTest.getError())).booleanValue();
    err = ((Boolean) server2.invoke(() -> ClientToServerDeltaDUnitTest.getError())).booleanValue();
    err = ((Boolean) client2.invoke(() -> ClientToServerDeltaDUnitTest.getError())).booleanValue();
    assertFalse("validation fails", err);
  }

  @Test
  public void testClientDeltaPropogationPutFetchesTheLatestValueWhenClientVersionIsOlder()
      throws Exception {
    // client did not register interest
    Integer PORT1 = ((Integer) server.invoke(() -> ClientToServerDeltaDUnitTest
        .createServerCache(Boolean.TRUE, Boolean.FALSE, Boolean.TRUE, Boolean.TRUE))).intValue();

    ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server.getHost()), new Integer(PORT1), Boolean.FALSE,
        Boolean.FALSE, Boolean.FALSE, null, Boolean.FALSE);
    Region r = cache.getRegion(REGION_NAME);
    DeltaTestImpl val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
        new TestObjectWithIdentifier("0", 0));
    r.put(KEY1, val);

    server.invoke(() -> {
      Region region = cache.getRegion(REGION_NAME);
      DeltaTestImpl val1 = (DeltaTestImpl) region.get(KEY1);
      val1.NEED_TO_RESET_T0_DELTA = false;
      val1.setIntVar(1);
      region.put(KEY1, val1);
    });

    DeltaTestImpl val2 = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
        new TestObjectWithIdentifier("0", 0));
    val2.setStr("1");
    val2.NEED_TO_RESET_T0_DELTA = false;
    r.put(KEY1, val2);

    server.invoke(() -> {
      Region region = cache.getRegion(REGION_NAME);
      assertThat((DeltaTestImpl) region.get(KEY1)).isNotNull();
    });

    DeltaTestImpl expected = new DeltaTestImpl(1, "1", new Double(0), new byte[0],
        new TestObjectWithIdentifier("0", 0));

    server.invoke(() -> {
      Region region = cache.getRegion(REGION_NAME);
      GeodeAwaitility.await()
          .untilAsserted(() -> assertThat((DeltaTestImpl) region.get(KEY1)).isEqualTo(expected));
    });

    GeodeAwaitility.await()
        .untilAsserted(() -> assertThat((DeltaTestImpl) r.get(KEY1)).isEqualTo(expected));
  }

  @Test
  public void clientDeltaPutFetchesTheLatestVersionIfNotYetReceivedQueuedEvent() {
    server.invoke(() -> setSlowStartForTesting());
    server2.invoke(() -> setSlowStartForTesting());
    initialise(false);

    DeltaTestImpl original = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
        new TestObjectWithIdentifier("0", 0));
    client.invoke(() -> {
      Region r = cache.getRegion(REGION_NAME);
      original.NEED_TO_RESET_T0_DELTA = false;
      r.put(KEY1, original);
    });

    client2.invoke(() -> {
      Region r = cache.getRegion(REGION_NAME);
      assertThat(r.get(KEY1)).isEqualTo(original);
    });

    client.invoke(() -> {
      Region r = cache.getRegion(REGION_NAME);
      DeltaTestImpl val = (DeltaTestImpl) r.get(KEY1);
      assertThat(val).isEqualTo(original);
      val.NEED_TO_RESET_T0_DELTA = false;
      val.setIntVar(1);
      r.put(KEY1, val);
    });

    client2.invoke(() -> {
      Region r = cache.getRegion(REGION_NAME);
      DeltaTestImpl val = (DeltaTestImpl) r.get(KEY1);
      // delta update should not arrive yet due to slow dispatcher
      assertThat(val).isEqualTo(original);
      val.NEED_TO_RESET_T0_DELTA = false;
      val.setStr("1");
      r.put(KEY1, val);
      Object o = r.get(KEY1);
      logger.info("object is " + o);
    });

    server.invoke(() -> {
      Region r = cache.getRegion(REGION_NAME);
      r.get(KEY1);
    });

    server2.invoke(() -> {
      Region r = cache.getRegion(REGION_NAME);
      r.get(KEY1);
    });

    DeltaTestImpl expected = new DeltaTestImpl(1, "1", new Double(0), new byte[0],
        new TestObjectWithIdentifier("0", 0));

    client.invoke(() -> {
      Region r = cache.getRegion(REGION_NAME);
      GeodeAwaitility.await()
          .untilAsserted(() -> assertThat((DeltaTestImpl) r.get(KEY1)).isEqualTo(expected));
    });

    client2.invoke(() -> {
      Region r = cache.getRegion(REGION_NAME);
      GeodeAwaitility.await()
          .untilAsserted(() -> assertThat((DeltaTestImpl) r.get(KEY1)).isEqualTo(expected));
    });
  }

  private void setSlowStartForTesting() {
    CacheClientProxy.isSlowStartForTesting = true;
    System.setProperty("slowStartTimeForTesting", "5000");
  }

  private static void putDeltaForCQ(String key, Integer numOfPuts, Integer[] cqIndices,
      Boolean[] satisfyQuery) {
    Region region = cache.getRegion(REGION_NAME);
    DeltaTestImpl val = null;
    for (int j = 0; j < numOfPuts; j++) {
      val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
          new TestObjectWithIdentifier("0", 0));
      for (int i = 0; i < cqIndices.length; i++) {
        switch (i) {
          case 0:
            val.setStr("CASE_0");
            // select *
            break;
          case 1:
            val.setStr("CASE_1");
            // select where intVar = 0
            if (satisfyQuery[i]) {
              val.setIntVar(0);
            } else {
              val.setIntVar(100);
            }
            break;
          case 2:
            val.setStr("CASE_2");
            // select where intVar > 0
            if (satisfyQuery[i]) {
              val.setIntVar(100);
            } else {
              val.setIntVar(-100);
            }
            break;
          case 3:
            val.setStr("CASE_3");
            // select where intVar < 0
            if (satisfyQuery[i]) {
              val.setIntVar(-100);
            } else {
              val.setIntVar(100);
            }
            break;
          default:
            break;
        }
      }
      region.put(key, val);
    }
  }

  /*
   * put delta ; not previous deltas
   */
  private static void putDelta(String key) {
    Region r = cache.getRegion(REGION_NAME);
    DeltaTestImpl val = null;
    for (int i = 0; i < NO_PUT_OPERATION; i++) {
      val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
          new TestObjectWithIdentifier("0", 0));
      switch (i) {
        case 1:
          val.setStr((String) putDelta[0]);
          break;
        case 2:
          val.setIntVar(((Integer) putDelta[2]).intValue());
          break;
      }
      r.put(key, val);
    }
  }

  private static void putLastKey() {
    Region r = cache.getRegion(REGION_NAME);
    r.put(LAST_KEY, LAST_KEY);
  }

  private static void putLastKeyWithDelta() {
    Region r = cache.getRegion(REGION_NAME);
    r.put(LAST_KEY, new DeltaTestImpl());
  }

  private static void setFirstSecondUpdate(Object first, Object second) {
    firstUpdate = first;
    secondUpdate = second;
  }

  private static Boolean getError() {
    return error;
  }

  /*
   * put delta full cycle
   */
  private static void put(String key) {
    Region r = cache.getRegion(REGION_NAME);
    DeltaTestImpl val = null;
    for (int i = 0; i < NO_PUT_OPERATION; i++) {
      switch (i) {
        case 0:
          val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
              new TestObjectWithIdentifier("0", 0));
          break;
        case 1:
          val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
              new TestObjectWithIdentifier("0", 0));
          val.setStr((String) putDelta[0]);
          break;
        case 2:
          val = new DeltaTestImpl(0, (String) putDelta[1], new Double(0), new byte[0],
              new TestObjectWithIdentifier("0", 0));
          val.setIntVar(((Integer) putDelta[2]).intValue());
          break;
      }
      r.put(key, val);
    }
  }

  /*
   * put delta with some times fromDelta fails to apply client sends back full object
   */
  private static void putWithFromDeltaERR(String key) {
    Region r = cache.getRegion(REGION_NAME);
    DeltaTestImpl val = null;
    for (int i = 0; i < NO_PUT_OPERATION; i++) {
      switch (i) {
        case 0:
          val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
              new TestObjectWithIdentifier("0", 0));
          break;
        case 1:
          val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
              new TestObjectWithIdentifier("0", 0));
          val.setStr((String) putDelta[1]);
          break;
        case 2:
          val = new DeltaTestImpl(0, (String) putDelta[1], new Double(0), new byte[0],
              new TestObjectWithIdentifier("0", 0));
          val.setIntVar(((Integer) putDelta[2]).intValue());
          break;
      }
      r.put(key, val);
    }
  }

  /*
   * put delta with some times toDelta fails to apply; client sends back full object
   */
  private static void putWithTODeltaERR(String key) {
    Region r = cache.getRegion(REGION_NAME);
    DeltaTestImpl val = null;
    for (int i = 0; i < NO_PUT_OPERATION; i++) {
      switch (i) {
        case 0:
          val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
              new TestObjectWithIdentifier("0", 0));
          break;
        case 1:
          val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
              new TestObjectWithIdentifier("0", 0));
          val.setIntVar(((Integer) putDelta[3]).intValue());
          break;
        case 2:
          val = new DeltaTestImpl(((Integer) putDelta[3]).intValue(), "0", new Double(0),
              new byte[0], new TestObjectWithIdentifier("0", 0));
          val.setStr((String) putDelta[0]);
          break;
      }
      r.put(key, val);
    }
  }

  public static Integer createServerCache(Boolean attachListener, Boolean isEmpty)
      throws Exception {
    return createServerCache(attachListener, isEmpty, Boolean.TRUE/* clone */,
        Boolean.TRUE/* enable delta */);
  }

  /*
   * create server cache
   */
  public static Integer createServerCache(Boolean attachListener, Boolean isEmpty, Boolean clone,
      Boolean enableDelta) throws Exception {
    // for validation
    updates = 0;
    create = 0;
    firstUpdate = null;
    secondUpdate = null;
    error = false;
    Properties props = new Properties();
    props.setProperty(DELTA_PROPAGATION, enableDelta.toString());
    new ClientToServerDeltaDUnitTest().createCache(props);
    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setConcurrencyChecksEnabled(true);
    if (isEmpty) {
      factory.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.ALL));
      factory.setDataPolicy(DataPolicy.EMPTY);

    } else {
      factory.setDataPolicy(DataPolicy.REPLICATE);
    }
    factory.setCloningEnabled(clone);
    RegionAttributes attrs = factory.create();
    region = cache.createRegion(REGION_NAME, attrs);

    AttributesMutator am = region.getAttributesMutator();
    if (attachListener) {
      am.addCacheListener(new CacheListenerAdapter() {
        @Override
        public void afterCreate(EntryEvent event) {
          create++;
        }

        @Override
        public void afterUpdate(EntryEvent event) {
          switch (updates) {
            case 0:
              // first delta
              validateUpdates(event, firstUpdate, "FIRST");
              updates++;
              break;
            case 1:
              // combine delta
              validateUpdates(event, firstUpdate, "FIRST");
              validateUpdates(event, secondUpdate, "SECOND");
              updates++;
              break;
            default:
              break;
          }
        }
      });
    } else if (!isEmpty) {
      am.addCacheListener(new CacheListenerAdapter() {
        @Override
        public void afterCreate(EntryEvent event) {
          switch (create) {
            case 1:
              validateUpdates(event, firstUpdate, "FIRST");
              create++;
              break;
            case 2:
              validateUpdates(event, secondUpdate, "SECOND");
              create++;
              break;
            default:
              create++;
              break;
          }
        }
        /*
         * public void afterUpdate(EntryEvent event) { updates++; }
         */
      });

    }

    CacheServer server = cache.addCacheServer();
    int port = getRandomAvailableTCPPort();
    server.setPort(port);
    // ensures updates to be sent instead of invalidations
    server.setNotifyBySubscription(true);
    server.start();
    return new Integer(server.getPort());
  }

  public static void createClientCache(String host, Integer port, Boolean attachListener,
      Boolean isEmpty, Boolean isCq) throws Exception {
    createClientCache(host, port, attachListener, isEmpty, isCq, new String[0], true);
  }

  public static void createClientCache(String host, Integer port, Boolean attachListener,
      Boolean isEmpty, Boolean isCq, String[] cqQueryString) throws Exception {
    createClientCache(host, port, attachListener, isEmpty, isCq, cqQueryString, true);
  }

  public static void createClientCache(String host, Integer port, Boolean attachListener,
      Boolean isEmpty, Boolean isCq, String[] cqQueryString, Boolean registerInterestAll)
      throws Exception {
    createClientCache(host, port, attachListener, isEmpty, isCq, cqQueryString, registerInterestAll,
        true);
  }

  /*
   * create client cache
   */
  public static void createClientCache(String host, Integer port, Boolean attachListener,
      Boolean isEmpty, Boolean isCq, String[] cqQueryString, Boolean registerInterestAll,
      Boolean enableSubscription) throws Exception {
    updates = 0;
    create = 0;
    firstUpdate = null;
    secondUpdate = null;
    error = false;
    lastKeyReceived = false;
    Properties props = new Properties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(LOCATORS, "");
    new ClientToServerDeltaDUnitTest().createCache(props);
    pool = (PoolImpl) PoolManager.createFactory().addServer(host, port.intValue())
        .setMinConnections(2)
        .setSubscriptionEnabled(enableSubscription).setSubscriptionRedundancy(0)
        .setReadTimeout(10000).setPingInterval(1000).setSocketBufferSize(32768)
        .create("ClientToServerDeltaDunitTestPool");
    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setConcurrencyChecksEnabled(true);
    if (isEmpty) {
      factory.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.ALL));
      factory.setDataPolicy(DataPolicy.EMPTY);
    } else {
      factory.setDataPolicy(DataPolicy.NORMAL);
    }
    factory.setPoolName(pool.getName());
    factory.setCloningEnabled(false);

    // region with empty data policy
    RegionAttributes attrs = factory.create();
    region = cache.createRegion(REGION_NAME, attrs);
    if (attachListener) {
      region.getAttributesMutator().addCacheListener(new CacheListenerAdapter() {
        @Override
        public void afterCreate(EntryEvent event) {
          create++;
          if (LAST_KEY.equals(event.getKey())) {
            lastKeyReceived = true;
          } ;
        }

        @Override
        public void afterUpdate(EntryEvent event) {
          switch (updates) {
            case 0:
              // first delta
              validateUpdates(event, firstUpdate, "FIRST");
              updates++;
              break;
            case 1:
              // combine delta
              validateUpdates(event, firstUpdate, "FIRST");
              validateUpdates(event, secondUpdate, "SECOND");
              updates++;
              break;
            default:
              break;
          }
        }
      });
    }
    if (registerInterestAll) {
      region.registerInterest("ALL_KEYS");
    }
    if (isCq) {
      CqAttributesFactory cqf = new CqAttributesFactory();
      CqListenerAdapter cqlist = new CqListenerAdapter() {
        @Override
        public void onEvent(CqEvent cqEvent) {
          Object key = cqEvent.getKey();
          if (LAST_KEY.equals(key)) {
            lastKeyReceived = true;
          }
          logger.fine(
              "CQ event received for (key, value): (" + key + ", " + cqEvent.getNewValue() + ")");
        }

        @Override
        public void onError(CqEvent cqEvent) {
          logger.fine("CQ error received for key: " + cqEvent.getKey());
        }
      };
      cqf.addCqListener(cqlist);
      CqAttributes cqa = cqf.create();
      for (int i = 0; i < cqQueryString.length; i++) {
        CqQuery cq = cache.getQueryService().newCq("Delta_Query_" + i, cqQueryString[i], cqa);
        cq.execute();
      }
    }
  }

  /*
   * create cache with properties
   */
  private void createCache(Properties props) throws Exception {
    DistributedSystem ds = getSystem(props);
    cache = CacheFactory.create(ds);
    assertNotNull(cache);
    logger = cache.getLogger();
  }

  // to validate updates in listener
  private static void validateUpdates(EntryEvent event, Object obj, String str) {
    if (obj instanceof String)
      assertTrue(str + " update missed ",
          ((DeltaTestImpl) event.getNewValue()).getStr().equals((String) obj));
    else if (obj instanceof Integer)
      assertTrue(str + " update missed ",
          ((DeltaTestImpl) event.getNewValue()).getIntVar() == (Integer) obj);
    else
      error = true;
  }

  /*
   * close cache
   */
  public static void closeCache() {
    if (cache != null && !cache.isClosed()) {
      cache.close();
      cache.getDistributedSystem().disconnect();
      lastKeyReceived = false;
    }
  }

  /**
   * This test does the following for single key:<br>
   * 1)Verifies that cacheless client calls toDelta <br>
   */
  @Test
  public void testEmptyClientAsFeederToServer() {
    Integer PORT1 = ((Integer) server
        .invoke(() -> ClientToServerDeltaDUnitTest.createServerCache(Boolean.FALSE, Boolean.FALSE)))
            .intValue();

    server2
        .invoke(() -> ClientToServerDeltaDUnitTest.createServerCache(Boolean.FALSE, Boolean.FALSE));

    client.invoke(() -> ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server.getHost()), new Integer(PORT1), Boolean.FALSE,
        Boolean.TRUE, Boolean.FALSE));

    /*
     * server2.invoke(() -> ClientToServerDeltaDUnitTest.setFirstSecondUpdate( putDelta[0],
     * (Integer)putDelta[2] ));
     */

    client.invoke(() -> ClientToServerDeltaDUnitTest.putDelta(KEY1));

    client.invoke(() -> ClientToServerDeltaDUnitTest.checkTodeltaCounter(new Integer(2)));
    server.invoke(() -> ClientToServerDeltaDUnitTest.checkFromdeltaCounter());
    server.invoke(() -> ClientToServerDeltaDUnitTest.checkTodeltaCounter(new Integer(0)));
    server2.invoke(() -> ClientToServerDeltaDUnitTest.checkFromdeltaCounter());
    server2.invoke(() -> ClientToServerDeltaDUnitTest.checkTodeltaCounter(new Integer(0)));
  }

  /**
   * This test does the following for single key:<br>
   * 1)Verifies that from delta should not called on server with empty data policy just by passed
   * delta to data store<br>
   */
  @Test
  public void testEmptyServerAsFeederToPeer() {
    Integer PORT1 = ((Integer) server
        .invoke(() -> ClientToServerDeltaDUnitTest.createServerCache(Boolean.FALSE, Boolean.TRUE)))
            .intValue();

    server2
        .invoke(() -> ClientToServerDeltaDUnitTest.createServerCache(Boolean.FALSE, Boolean.FALSE));

    client.invoke(() -> ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server.getHost()), new Integer(PORT1), Boolean.FALSE,
        Boolean.TRUE, Boolean.FALSE));

    client.invoke(() -> ClientToServerDeltaDUnitTest.putDelta(KEY1));

    client.invoke(() -> ClientToServerDeltaDUnitTest.checkTodeltaCounter(new Integer(2)));

    server.invoke(() -> ClientToServerDeltaDUnitTest.checkDeltaFeatureNotUsed());

    server2.invoke(() -> ClientToServerDeltaDUnitTest.checkTodeltaCounter(new Integer(0)));
    server2.invoke(() -> ClientToServerDeltaDUnitTest.checkFromdeltaCounter());
  }

  /**
   * This test does verifies that server with empty data policy sends deltas to the client which can
   * handle deltas. Server sends full values which can not handle deltas.
   */
  @Test
  public void testClientsConnectedToEmptyServer() {
    Integer PORT1 = ((Integer) server
        .invoke(() -> ClientToServerDeltaDUnitTest.createServerCache(Boolean.FALSE, Boolean.TRUE)))
            .intValue();

    server2
        .invoke(() -> ClientToServerDeltaDUnitTest.createServerCache(Boolean.FALSE, Boolean.FALSE));

    client.invoke(() -> ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server.getHost()), new Integer(PORT1), Boolean.TRUE,
        Boolean.TRUE, Boolean.FALSE));

    client2.invoke(() -> ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server.getHost()), new Integer(PORT1), Boolean.TRUE,
        Boolean.FALSE, Boolean.FALSE));

    int deltaSent =
        (Integer) server2.invoke(() -> ClientToServerDeltaDUnitTest.putsWhichReturnsDeltaSent());

    client.invoke(() -> ClientToServerDeltaDUnitTest.waitForLastKey());
    client.invoke(() -> ClientToServerDeltaDUnitTest.checkForDelta());

    client2.invoke(() -> ClientToServerDeltaDUnitTest.waitForLastKey());
    client2.invoke(() -> ClientToServerDeltaDUnitTest.checkDeltaInvoked(new Integer(deltaSent)));
  }

  /**
   * This test does the following for single key:<br>
   * 1)Verifies that To delta called on client should be equal to fromDeltaCounter on datastore <br>
   */
  @Test
  public void testClientNonEmptyEmptyServerAsFeederToPeer() {
    Integer PORT1 = ((Integer) server
        .invoke(() -> ClientToServerDeltaDUnitTest.createServerCache(Boolean.FALSE, Boolean.TRUE)))
            .intValue();

    server2
        .invoke(() -> ClientToServerDeltaDUnitTest.createServerCache(Boolean.FALSE, Boolean.FALSE));

    client.invoke(() -> ClientToServerDeltaDUnitTest.createClientCache(
        NetworkUtils.getServerHostName(server.getHost()), new Integer(PORT1), Boolean.FALSE,
        Boolean.FALSE, Boolean.FALSE));

    client.invoke(() -> ClientToServerDeltaDUnitTest.putDelta(KEY1));

    client.invoke(() -> ClientToServerDeltaDUnitTest.checkTodeltaCounter(new Integer(2)));

    server.invoke(() -> ClientToServerDeltaDUnitTest.checkDeltaFeatureNotUsed());

    server2.invoke(() -> ClientToServerDeltaDUnitTest.checkTodeltaCounter(new Integer(0)));
    server2.invoke(() -> ClientToServerDeltaDUnitTest.checkFromdeltaCounter());
  }

  /**
   * This test has client1 connected to server1 and client2 connected to server2. client2 has a CQ
   * registered on the server2. client1 creates a key on server1 which satisfies the CQ and it then
   * updates the same key with a value so that it sould not satisfy the CQ. The cloning is set to
   * false. The test ensures that the client2 gets the update event. This implies that server2
   * clones the old value prior to applying delta received from server1 and then processes the CQ.
   */
  @Test
  public void testC2CDeltaPropagationWithCQ() throws Exception {
    initialise(false/* clone */, new String[] {CQs[1]}, true/* CQ */, true/* RI */,
        true/* enable delta */);
    client.invoke(ClientToServerDeltaDUnitTest.class, "createKeys",
        new Object[] {new String[] {KEY1}});
    client.invoke(ClientToServerDeltaDUnitTest.class, "putDeltaForCQ", new Object[] {KEY1,
        new Integer(1), new Integer[] {1}, new Boolean[] {false, true, false, false}});

    client.invoke(() -> ClientToServerDeltaDUnitTest.putLastKeyWithDelta());

    client2.invoke(() -> ClientToServerDeltaDUnitTest.waitForLastKey());
    client2.invoke(() -> ClientToServerDeltaDUnitTest.verifyDeltaReceived());
  }

  /**
   * This test ensures that a server sends delta bytes to a client even if that client is not
   * interested in that event but is getting the event only because the event satisfies a CQ which
   * the client has registered with the server.
   */
  @Test
  public void testC2CDeltaPropagationWithCQWithoutRI() throws Exception {
    initialise(false/* clone */, new String[] {CQs[1]}, true/* CQ */, false/* RI */,
        true/* enable delta */);

    client.invoke(ClientToServerDeltaDUnitTest.class, "createKeys",
        new Object[] {new String[] {KEY1}});
    client.invoke(ClientToServerDeltaDUnitTest.class, "putDeltaForCQ", new Object[] {KEY1,
        new Integer(1), new Integer[] {1}, new Boolean[] {false, true, false, false}});

    client.invoke(() -> ClientToServerDeltaDUnitTest.putLastKeyWithDelta());

    client2.invoke(() -> ClientToServerDeltaDUnitTest.waitForLastKey());
    server2.invoke(() -> ClientToServerDeltaDUnitTest.verifyDeltaSent(Integer.valueOf(1)));
  }

  @Test
  public void testClientSendsFullValueToServerWhenDeltaOffAtServer() {
    initialise2(false/* clone */, new String[] {CQs[1]}, false/* CQ */, true/* RI */,
        false/* enable delta */);
    // set expected value on server
    server.invoke(() -> ClientToServerDeltaDUnitTest.setFirstSecondUpdate(putDelta[0],
        (Integer) putDelta[2]));

    client.invoke(() -> ClientToServerDeltaDUnitTest.putDelta(DELTA_KEY));

    assertFalse("Delta Propagation feature used at client.",
        (Boolean) client.invoke(() -> DeltaTestImpl.toDeltaFeatureUsed()));
    assertFalse("Delta Propagation feature used at server.",
        (Boolean) server.invoke(() -> DeltaTestImpl.fromDeltaFeatureUsed()));
    assertFalse(
        "Failures at client while calculating delta. But delta-propagation is false at server.",
        ((Boolean) client.invoke(() -> DeltaTestImpl.isToDeltaFailure())).booleanValue());
    assertFalse(
        "Failures at server while applying delta. But delta-propagation is false at server.",
        (Boolean) server.invoke(() -> DeltaTestImpl.isFromDeltaFailure()));
  }

  @Test
  public void testC2SDeltaPropagationWithOldValueInvalidatedAtServer() throws Exception {
    String key = "DELTA_KEY";
    Integer port1 = (Integer) server
        .invoke(() -> ClientToServerDeltaDUnitTest.createServerCache(false, false, false, true));
    createClientCache("localhost", port1, false, false, false, null, false, false);

    LocalRegion region = (LocalRegion) cache.getRegion(REGION_NAME);
    region.put(key, new DeltaTestImpl());

    server.invoke(() -> ClientToServerDeltaDUnitTest.doInvalidate(key));
    DeltaTestImpl value = new DeltaTestImpl();
    value.setStr("UPDATED_VALUE");
    region.put(key, value);

    assertTrue(region.getCachePerfStats().getDeltasSent() == 1);
    assertTrue(region.getCachePerfStats().getDeltaFullValuesSent() == 1);
  }

  public static void doInvalidate(String key) {
    Region region = cache.getRegion(REGION_NAME);
    region.invalidate(key);
  }

  public static void verifyDeltaReceived() {
    assertTrue(
        "Expected 1 fromDelta() invokation but was " + DeltaTestImpl.getFromDeltaInvokations(),
        DeltaTestImpl.getFromDeltaInvokations() == 1);
  }

  public static void verifyDeltaSent(Integer deltas) {
    CacheClientNotifier ccn = ((CacheServerImpl) cache.getCacheServers().toArray()[0]).getAcceptor()
        .getCacheClientNotifier();

    int numOfDeltasSent = ((CacheClientProxy) ccn.getClientProxies().toArray()[0]).getStatistics()
        .getDeltaMessagesSent();
    assertTrue("Expected " + deltas + " deltas to be sent but " + numOfDeltasSent + " were sent.",
        numOfDeltasSent == deltas);
  }

  public static void checkFromdeltaCounter() {
    assertTrue(
        "FromDelta counters do not match, expected: " + (NO_PUT_OPERATION - 1) + ", actual: "
            + DeltaTestImpl.getFromDeltaInvokations(),
        DeltaTestImpl.getFromDeltaInvokations() >= (NO_PUT_OPERATION - 1));
  }

  public static void checkTodeltaCounter(Integer count) {
    assertTrue(
        "ToDelta counters do not match, expected: " + count.intValue() + ", actual: "
            + DeltaTestImpl.getToDeltaInvokations(),
        DeltaTestImpl.getToDeltaInvokations() >= count.intValue());
  }

  public static void checkDeltaFeatureNotUsed() {
    assertTrue("Delta Propagation feature used.", !(DeltaTestImpl.deltaFeatureUsed()));
  }

  public static void createKeys(String[] keys) {
    Region region = cache.getRegion(REGION_NAME);
    for (int i = 0; i < keys.length; i++) {
      region.create(keys[i], new DeltaTestImpl());
    }
  }

  public static int putsWhichReturnsDeltaSent() throws Exception {
    csDelta = new CSDeltaTestImpl();
    region.put(DELTA_KEY, csDelta);
    for (int i = 0; i < NO_PUT_OPERATION; i++) {
      csDelta.setIntVar(i);
      region.put(DELTA_KEY, csDelta);
    }
    region.put(LAST_KEY, "");

    return csDelta.deltaSent;
  }

  public static void checkDeltaInvoked(Integer deltaSent) {
    assertTrue(
        "Delta applied :" + ((CSDeltaTestImpl) region.get(DELTA_KEY)).getDeltaApplied()
            + "\n Delta sent :" + deltaSent,
        ((CSDeltaTestImpl) region.get(DELTA_KEY)).getDeltaApplied() == deltaSent);
  }

  public static void checkForDelta() {
    assertTrue("Delta sent to EMPTY data policy region",
        DeltaTestImpl.getFromDeltaInvokations() == 0);
  }

  public static void waitForLastKey() {
    WaitCriterion wc = new WaitCriterion() {
      @Override
      public boolean done() {
        return ClientToServerDeltaDUnitTest.lastKeyReceived;
      }

      @Override
      public String description() {
        return "Last key NOT received.";
      }
    };
    GeodeAwaitility.await().untilAsserted(wc);
  }

  static class CSDeltaTestImpl extends DeltaTestImpl {

    int deltaSent = 0;
    int deltaApplied = 0;

    public CSDeltaTestImpl() {}

    @Override
    public void toDelta(DataOutput out) throws IOException {
      super.toDelta(out);
      deltaSent++;
    }

    @Override
    public void fromDelta(DataInput in) throws IOException {
      super.fromDelta(in);
      deltaApplied++;
    }

    public int getDeltaSent() {
      return deltaSent;
    }

    public int getDeltaApplied() {
      return deltaApplied;
    }

    @Override
    public String toString() {
      return "CSDeltaTestImpl[deltaApplied=" + deltaApplied + "]" + super.toString();
    }
  }

}