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

import static org.apache.geode.distributed.ConfigurationProperties.CONSERVE_SOCKETS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
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.CacheFactory;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionFactory;
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.server.CacheServer;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.DistributionMessageObserver;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.cache.DistributedCacheOperation;
import org.apache.geode.internal.cache.DistributedPutAllOperation;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.RegionEntry;
import org.apache.geode.internal.cache.tier.sockets.BaseCommand;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.command.Put70;
import org.apache.geode.internal.cache.versions.VMVersionTag;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
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.junit.categories.SerializationTest;
import org.apache.geode.util.internal.GeodeGlossary;

@Category({SerializationTest.class})
public class ClientsWithVersioningRetryDUnitTest extends JUnit4CacheTestCase {

  // list of expected exceptions to remove in tearDown2()
  static List<IgnoredException> expectedExceptions = new LinkedList<IgnoredException>();

  @Override
  public final void postSetUp() throws Exception {
    Invoke.invokeInEveryVM(new SerializableRunnable() {
      @Override
      public void run() {
        // Disable endpoint shuffling, so that the client will always connect
        // to the first server we give it.
        System.setProperty(GeodeGlossary.GEMFIRE_PREFIX + "bridge.disableShufflingOfEndpoints",
            "true");
      }

    });
  }

  @Override
  public final void postTearDownCacheTestCase() throws Exception {
    Invoke.invokeInEveryVM(new SerializableRunnable() {
      @Override
      public void run() {
        System.setProperty(GeodeGlossary.GEMFIRE_PREFIX + "bridge.disableShufflingOfEndpoints",
            "false");
      }
    });
    for (IgnoredException ex : expectedExceptions) {
      ex.remove();
    }
  }

  /**
   * Test that we can successfully retry a distributed put all and get the version information.
   * second failure in bug 44951
   */
  @Test
  public void testRetryPut() {
    Host host = Host.getHost(0);
    final VM vm0 = host.getVM(0);
    final VM vm1 = host.getVM(1);

    createServerRegion(vm0, RegionShortcut.REPLICATE);
    createServerRegion(vm1, RegionShortcut.REPLICATE);

    // create an event tag in vm0 and then replay that event in vm1
    final DistributedMember memberID =
        (DistributedMember) vm0.invoke(new SerializableCallable("get id") {
          @Override
          public Object call() {
            return ((DistributedRegion) getCache().getRegion("region")).getDistributionManager()
                .getDistributionManagerId();
          }
        });
    vm0.invoke(new SerializableCallable("create entry with fake event ID") {
      @Override
      public Object call() {
        DistributedRegion dr = (DistributedRegion) getCache().getRegion("region");
        VersionTag tag = new VMVersionTag();
        tag.setMemberID(dr.getVersionMember());
        tag.setRegionVersion(123);
        tag.setEntryVersion(9);
        tag.setVersionTimeStamp(System.currentTimeMillis());
        EventID eventID = new EventID(new byte[0], 1, 0);
        EntryEventImpl event = EntryEventImpl.create(dr, Operation.CREATE, "TestObject",
            "TestValue", null, false, memberID, true, eventID);
        event.setVersionTag(tag);
        event.setContext(new ClientProxyMembershipID(memberID));
        dr.recordEvent(event);
        event.release();
        return memberID;
      }
    });
    vm1.invoke(new SerializableRunnable("recover event tag in vm1 from vm0") {
      @Override
      public void run() {
        DistributedRegion dr = (DistributedRegion) getCache().getRegion("region");
        EventID eventID = new EventID(new byte[0], 1, 0);
        EntryEventImpl event = EntryEventImpl.create(dr, Operation.CREATE, "TestObject",
            "TestValue", null, false, memberID, true, eventID);
        try {
          event.setContext(new ClientProxyMembershipID(memberID));
          boolean recovered =
              ((BaseCommand) Put70.getCommand()).recoverVersionTagForRetriedOperation(event);
          assertTrue("Expected to recover the version for this event ID",
              recovered);
          assertEquals("Expected the region version to be 123", 123,
              event.getVersionTag().getRegionVersion());
        } finally {
          event.release();
        }
      }
    });
    // bug #48205 - a retried op in PR nodes not owning the primary bucket
    // may already have a version assigned to it in another backup bucket
    vm1.invoke(new SerializableRunnable("recover posdup event tag in vm1 event tracker from vm0") {
      @Override
      public void run() {
        DistributedRegion dr = (DistributedRegion) getCache()
            .getRegion("region");
        EventID eventID = new EventID(new byte[0], 1, 0);
        EntryEventImpl event = EntryEventImpl
            .create(dr, Operation.CREATE, "TestObject",
                "TestValue", null, false, memberID, true, eventID);
        event.setPossibleDuplicate(true);
        try {
          dr.hasSeenEvent(event);
          assertTrue("Expected to recover the version for the event ID",
              event.getVersionTag() != null);
        } finally {
          event.release();
        }
      }
    });
  }

  enum OpType {
    CREATE,
    PUT,
    DESTROY,
    INVALIDATE,
    PUT_IF_ABSENT,
    REPLACE,
    REPLACE_WITH_OLDVALUE,
    REMOVE
  };

  @Test
  public void testRetriedCreate() {
    doRetriedTest(OpType.CREATE);
  }

  @Test
  public void testRetriedPut() {
    doRetriedTest(OpType.PUT);
  }

  @Test
  public void testRetriedDestroy() {
    doRetriedTest(OpType.DESTROY);
  }

  @Test
  public void testRetriedInvalidate() {
    doRetriedTest(OpType.INVALIDATE);
  }

  @Test
  public void testRetriedPutIfAbsent() {
    doRetriedTest(OpType.PUT_IF_ABSENT);
  }

  @Test
  public void testRetriedReplace() {
    doRetriedTest(OpType.REPLACE);
  }

  @Test
  public void testRetriedReplaceWithOldValue() {
    doRetriedTest(OpType.REPLACE_WITH_OLDVALUE);
  }

  @Test
  public void testRetriedRemove() {
    doRetriedTest(OpType.REMOVE);
  }

  private void doRetriedTest(final OpType opType) {
    Host host = Host.getHost(0);
    final VM vm0 = host.getVM(0);
    final VM vm1 = host.getVM(1);
    final VM vm3 = host.getVM(3);

    int port0 = createServerRegion(vm0, RegionShortcut.REPLICATE);
    int port1 = createServerRegion(vm1, RegionShortcut.REPLICATE);
    createClientRegion(vm3, port0, port1);

    vm0.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        Region region = getCache().getRegion("region");
        if (opType != OpType.CREATE && opType != OpType.PUT_IF_ABSENT) {
          region.put(0, "value");
        }

        // Add a listener to close vm0 when we send a distributed operation
        // this will cause a retry after we have applied the original operation to
        // the cache, causing a retry
        DistributionMessageObserver
            .setInstance(new DistributionMessageObserver() {

              @Override
              public void beforeSendMessage(ClusterDistributionManager dm,
                  DistributionMessage message) {
                if (message instanceof DistributedCacheOperation.CacheOperationMessage) {
                  DistributedCacheOperation.CacheOperationMessage com =
                      (DistributedCacheOperation.CacheOperationMessage) message;
                  VersionTag tag = com.getVersionTag();
                  if (((opType == OpType.CREATE || opType == OpType.PUT_IF_ABSENT)
                      && tag.getEntryVersion() == 1) || tag.getEntryVersion() == 2) {
                    DistributionMessageObserver.setInstance(null);
                    disconnectFromDS(vm0);
                  }
                }
              }
            });

      }
    });

    // this operation will trigger vm0 to be closed, and the operation will be retried
    vm3.invoke(new SerializableCallable("perform update in client") {
      @Override
      public Object call() throws Exception {
        Region region = getCache().getRegion("region");
        switch (opType) {
          case CREATE:
            region.create(0, "newvalue");
            break;
          case PUT:
            region.put(0, "newvalue");
            break;
          case DESTROY:
            region.destroy(0);
            break;
          case INVALIDATE:
            region.invalidate(0);
            break;
          case PUT_IF_ABSENT:
            region.putIfAbsent(0, "newvalue");
            break;
          case REPLACE:
            region.replace(0, "newvalue");
            break;
          case REPLACE_WITH_OLDVALUE:
            region.replace(0, "value", "newvalue");
            break;
          case REMOVE:
            region.remove(0, "value");
            break;
        }
        return null;
      }
    });

    // Verify the observer was triggered
    vm1.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        // if the observer was triggered, it would have cleared itself
        assertNull(DistributionMessageObserver.getInstance());
        VersionTag tag = ((LocalRegion) getCache().getRegion("region"))
            .getVersionTag(0);
        if (opType == OpType.CREATE || opType == OpType.PUT_IF_ABSENT) {
          assertEquals(1, tag.getRegionVersion());
        } else {
          assertEquals(2, tag.getRegionVersion());
        }
      }
    });

    // Make sure vm0 did in fact shut down
    vm0.invoke(new SerializableRunnable() {
      @Override
      public void run() {
        GemFireCacheImpl cache = (GemFireCacheImpl) basicGetCache();
        assertTrue(cache == null || cache.isClosed());
      }
    });
  }

  /**
   * Test that we can successfully retry a distributed put all and get the version information. bug
   * #45059
   */
  @Test
  public void testRetryPutAll() {
    Host host = Host.getHost(0);
    final VM vm0 = host.getVM(0);
    final VM vm1 = host.getVM(1);
    final VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);

    createServerRegion(vm0, RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);
    vm0.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        // Make sure the bucket 0 is primary in this member.
        Region region = getCache().getRegion("region");
        region.put(0, "value");

        // Add a listener to close vm1 when we send a distributed put all operation
        // this will cause a retry after we have applied the original put all to
        // the cache, causing a retry
        DistributionMessageObserver
            .setInstance(new DistributionMessageObserver() {

              @Override
              public void beforeSendMessage(ClusterDistributionManager dm,
                  DistributionMessage message) {
                if (message instanceof DistributedPutAllOperation.PutAllMessage) {
                  DistributionMessageObserver.setInstance(null);
                  disconnectFromDS(vm1);
                }
              }
            });

      }
    });

    int port1 = createServerRegion(vm1,
        RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);
    int port2 = createServerRegion(vm2,
        RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);
    createClientRegion(vm3, port1, port2);

    // This will be a put all to bucket 0
    // Here's the expected sequence
    // client->vm1 (accessor0)
    // vm1->vm0
    // vm0 will kill vm1
    // vm0->vm2
    // client will retry the putall
    vm3.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region region = getCache().getRegion("region");
        Map map = new HashMap();
        map.put(0, "a");
        map.put(113, "b");
        region.putAll(map);
        RegionEntry entry = ((LocalRegion) region).getRegionEntry(0);
        assertNotNull(entry);
        assertNotNull(entry.getVersionStamp());
        assertEquals(2, entry.getVersionStamp().getEntryVersion());
        return null;
      }
    });

    // Verify the observer was triggered
    vm0.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        // if the observer was triggered, it would have cleared itself
        assertNull(DistributionMessageObserver.getInstance());
      }
    });

    // Make sure vm1 did in fact shut down
    vm1.invoke(new SerializableRunnable() {
      @Override
      public void run() {
        GemFireCacheImpl cache = (GemFireCacheImpl) basicGetCache();
        assertTrue(cache == null || cache.isClosed());
      }
    });
  }

  /**
   * Test that we can successfully retry a distributed putAll on an accessor and get the version
   * information. bug #48205
   */
  @Test
  public void testRetryPutAllInAccessor() {
    Host host = Host.getHost(0);
    final VM vm0 = host.getVM(0);
    final VM vm1 = host.getVM(1);
    final VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);

    LogWriterUtils.getLogWriter().info("creating region in vm0");
    createRegionInPeer(vm0, RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);

    vm0.invoke(new SerializableRunnable() {
      @Override
      public void run() {
        // Make sure the bucket 0 is primary in this member.
        Region region = getCache().getRegion("region");
        region.put(0, "value");
      }
    });

    LogWriterUtils.getLogWriter().info("creating region in vm1");
    createRegionInPeer(vm1, RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);
    LogWriterUtils.getLogWriter().info("creating region in vm2");
    createRegionInPeer(vm2, RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);
    LogWriterUtils.getLogWriter().info("creating region in vm3");
    createRegionInPeer(vm3, RegionShortcut.PARTITION_PROXY);

    expectedExceptions
        .add(IgnoredException.addIgnoredException("RuntimeException", vm2));
    vm2.invoke(
        new SerializableRunnable("install message listener to ignore update") {
          @Override
          public void run() {
            // Add a listener to close vm2 when we send a distributed put all operation
            // this will cause a retry after we have applied the original put all to
            // the cache, causing a retry
            DistributionMessageObserver
                .setInstance(new DistributionMessageObserver() {

                  @Override
                  public void beforeProcessMessage(ClusterDistributionManager dm,
                      DistributionMessage msg) {
                    if (msg instanceof DistributedPutAllOperation.PutAllMessage) {
                      DistributionMessageObserver.setInstance(null);
                      Wait.pause(
                          5000); // give vm1 time to process the message that we're ignoring
                      disconnectFromDS(vm0);
                      // no reply will be sent to vm0 due to this exception, but that's okay
                      // because vm0 has been shut down
                      throw new RuntimeException(
                          "test code is ignoring message: " + msg);
                    }
                  }
                });

          }
        });

    // This will be a put all to bucket 0
    // Here's the expected sequence
    // accessor->vm0 (primary)
    // vm0->vm1, vm2
    // vm2 will ignore the message & kill vm0
    // accessor->vm2 or vm1
    // version tag is recovered and put in the event & cache
    vm3.invoke(new SerializableCallable("perform putAll in accessor") {
      @Override
      public Object call() throws Exception {
        Region region = getCache().getRegion("region");
        Map map = new HashMap();
        map.put(0, "a");
        map.put(113, "b");
        region.putAll(map);
        return null;
      }
    });

    // verify that the version is correct
    vm1.invoke(new SerializableRunnable("verify vm1") {

      @Override
      public void run() {
        // if the observer was triggered, it would have cleared itself
        assertNull(DistributionMessageObserver.getInstance());

        Region region = getCache().getRegion("region");
        VersionTag tag = ((LocalRegion) region).getVersionTag(0);
        assertEquals(2, tag.getEntryVersion());
      }
    });

    // Verify the observer was triggered and the version is correct
    vm2.invoke(new SerializableRunnable("verify vm2") {

      @Override
      public void run() {
        // if the observer was triggered, it would have cleared itself
        assertNull(DistributionMessageObserver.getInstance());

        Region region = getCache().getRegion("region");
        VersionTag tag = ((LocalRegion) region).getVersionTag(0);
        assertEquals(2, tag.getEntryVersion());
      }
    });

    // Make sure vm1 did in fact shut down
    vm0.invoke(new SerializableRunnable() {
      @Override
      public void run() {
        GemFireCacheImpl cache = (GemFireCacheImpl) basicGetCache();
        assertTrue(cache == null || cache.isClosed());
      }
    });
  }

  private void disconnectFromDS(VM vm) {
    vm.invoke(new SerializableCallable("disconnecting vm " + vm) {
      @Override
      public Object call() throws Exception {
        disconnectFromDS();
        return null;
      }
    });
  }

  private int createServerRegion(VM vm, final RegionShortcut shortcut) {
    SerializableCallable createRegion = new SerializableCallable(
        "create server region") {
      @Override
      public Object call() throws Exception {
        RegionFactory<Object, Object> rf = getCache()
            .createRegionFactory(shortcut);
        if (!shortcut.equals(RegionShortcut.REPLICATE)) {
          rf.setPartitionAttributes(
              new PartitionAttributesFactory().setRedundantCopies(2).create());
        }
        rf.create("region");

        CacheServer server = getCache().addCacheServer();
        int port = AvailablePortHelper.getRandomAvailableTCPPort();
        server.setPort(port);
        server.start();
        return port;
      }
    };

    return (Integer) vm.invoke(createRegion);
  }

  private void createRegionInPeer(VM vm, final RegionShortcut shortcut) {
    SerializableCallable createRegion = new SerializableCallable(
        "create peer region") {
      @Override
      public Object call() throws Exception {
        RegionFactory<Object, Object> rf = getCache()
            .createRegionFactory(shortcut);
        if (!shortcut.equals(RegionShortcut.REPLICATE)) {
          rf.setPartitionAttributes(
              new PartitionAttributesFactory().setRedundantCopies(2).create());
        }
        rf.create("region");
        return null;
      }
    };
    vm.invoke(createRegion);
  }

  @Override
  public Properties getDistributedSystemProperties() {
    Properties p = super.getDistributedSystemProperties();
    p.put(CONSERVE_SOCKETS, "false");
    return p;
  }

  private int createServerRegionWithPersistence(VM vm,
      final boolean persistentPdxRegistry) {
    SerializableCallable createRegion = new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        CacheFactory cf = new CacheFactory();
        if (persistentPdxRegistry) {
          cf.setPdxPersistent(true).setPdxDiskStore("store");
        }
        //
        Cache cache = getCache(cf);
        cache.createDiskStoreFactory().setDiskDirs(getDiskDirs())
            .create("store");

        AttributesFactory af = new AttributesFactory();
        af.setScope(Scope.DISTRIBUTED_ACK);
        af.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
        af.setDiskStoreName("store");
        createRootRegion("testSimplePdx", af.create());

        CacheServer server = getCache().addCacheServer();
        int port = AvailablePortHelper.getRandomAvailableTCPPort();
        server.setPort(port);
        server.start();
        return port;
      }
    };

    return (Integer) vm.invoke(createRegion);
  }

  private int createServerAccessor(VM vm) {
    SerializableCallable createRegion = new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        AttributesFactory af = new AttributesFactory();
        af.setScope(Scope.DISTRIBUTED_ACK);
        af.setDataPolicy(DataPolicy.EMPTY);
        createRootRegion("testSimplePdx", af.create());

        CacheServer server = getCache().addCacheServer();
        int port = AvailablePortHelper.getRandomAvailableTCPPort();
        server.setPort(port);
        server.start();
        return port;
      }
    };

    return (Integer) vm.invoke(createRegion);
  }

  private void createClientRegion(final VM vm, final int port1,
      final int port2) {
    SerializableCallable createRegion = new SerializableCallable(
        "create client region in " + vm) {
      @Override
      public Object call() throws Exception {
        ClientCacheFactory cf = new ClientCacheFactory();

        cf.addPoolServer(NetworkUtils.getServerHostName(vm.getHost()), port1);
        cf.addPoolServer(NetworkUtils.getServerHostName(vm.getHost()), port2);
        cf.setPoolPRSingleHopEnabled(false);
        cf.setPoolReadTimeout(10 * 60 * 1000);
        cf.setPoolSubscriptionEnabled(true);

        ClientCache cache = getClientCache(cf);
        Region region =
            cache.createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY)
                .create("region");
        region.registerInterest("ALL_KEYS");
        return null;
      }
    };
    vm.invoke(createRegion);
  }
}