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

import static org.apache.geode.cache.RegionShortcut.PARTITION;
import static org.apache.geode.cache.RegionShortcut.REPLICATE;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import static org.assertj.core.api.Assertions.assertThat;

import java.util.Collection;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
import java.util.function.Function;

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

import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.RegionFunctionContext;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.LonerDistributionManager;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.execute.util.TypedFunctionService;
import org.apache.geode.test.junit.categories.FunctionServiceTest;

/**
 * This tests make sure that, in case of LonerDistributedSystem we don't get ClassCast Exception.
 * Just making sure that the function executed on lonerDistributedSystem.
 *
 * <p>
 * Regression test for bug: If invoked from LonerDistributedSystem, FunctionService.onMembers()
 * throws ClassCastException
 */
@Category(FunctionServiceTest.class)
@SuppressWarnings("serial")
public class FunctionExecutionOnLonerRegressionTest {

  private InternalCache cache;

  private Set<String> keysForGet;
  private Set<String> expectedValues;

  @Before
  public void setUp() {
    cache = (InternalCache) new CacheFactory(getDistributedSystemProperties()).create();
  }

  @After
  public void tearDown() {
    cache.close();
  }

  @Test
  public void precondition_isLonerDistributionManager() {
    DistributionManager distributionManager = cache.getDistributionManager();

    assertThat(distributionManager).isInstanceOf(LonerDistributionManager.class);
  }

  @Test
  public void executeFunctionOnLonerWithPartitionedRegionShouldNotThrowClassCastException() {
    Region<String, String> region = cache
        .<String, String>createRegionFactory(PARTITION)
        .create("region");

    populateRegion(region);

    ResultCollector<Collection<String>, Collection<String>> resultCollector = TypedFunctionService
        .<Void, Collection<String>, Collection<String>>onRegion(region)
        .withFilter(keysForGet)
        .execute(new TestFunction(DataSetSupplier.PARTITIONED));

    assertThat(resultCollector.getResult())
        .containsExactlyInAnyOrder(expectedValues.toArray(new String[0]));
  }

  @Test
  public void executeFunctionOnLonerWithReplicateRegionShouldNotThrowClassCastException() {
    Region<String, String> region = cache
        .<String, String>createRegionFactory(REPLICATE)
        .create("region");

    populateRegion(region);

    ResultCollector<Collection<String>, Collection<String>> resultCollector = TypedFunctionService
        .<Void, Collection<String>, Collection<String>>onRegion(region)
        .withFilter(keysForGet)
        .execute(new TestFunction(DataSetSupplier.REPLICATE));

    assertThat(resultCollector.getResult())
        .containsExactlyInAnyOrder(expectedValues.toArray(new String[0]));
  }

  private Properties getDistributedSystemProperties() {
    Properties config = new Properties();
    config.setProperty(MCAST_PORT, "0");
    config.setProperty(LOCATORS, "");
    config.setProperty(SERIALIZABLE_OBJECT_FILTER,
        "org.apache.geode.test.junit.rules.**;org.apache.geode.internal.cache.execute.**;org.apache.geode.internal.cache.functions.**;org.apache.geode.test.dunit.**");
    return config;
  }

  private void populateRegion(Region<String, String> region) {
    keysForGet = new HashSet<>();
    expectedValues = new HashSet<>();

    for (int i = 0; i < 20; i++) {
      String key = "KEY_" + i;
      String value = "VALUE_" + i;

      region.put(key, value);

      if (i == 4 || i == 7 || i == 9) {
        keysForGet.add(key);
        expectedValues.add(value);
      }
    }
  }

  private enum DataSetSupplier {
    PARTITIONED(PartitionRegionHelper::getLocalDataForContext),
    REPLICATE(RegionFunctionContext::getDataSet);

    private final Function<RegionFunctionContext, Region<String, String>> dataSet;

    DataSetSupplier(Function<RegionFunctionContext, Region<String, String>> dataSet) {

      this.dataSet = dataSet;
    }

    Region<String, String> dataSet(RegionFunctionContext context) {
      return dataSet.apply(context);
    }
  }

  private static class TestFunction implements org.apache.geode.cache.execute.Function<String> {

    private final DataSetSupplier dataSetSupplier;

    private TestFunction(DataSetSupplier dataSetSupplier) {
      this.dataSetSupplier = dataSetSupplier;
    }

    @Override
    public void execute(FunctionContext<String> context) {
      RegionFunctionContext regionFunctionContext = (RegionFunctionContext) context;
      Set<String> keys = uncheckedCast(regionFunctionContext.getFilter());
      String lastKey = keys.iterator().next();
      keys.remove(lastKey);

      Region<String, String> region = dataSetSupplier.dataSet(regionFunctionContext);

      for (String key : keys) {
        context.getResultSender().sendResult(region.get(key));
      }

      context.getResultSender().lastResult(region.get(lastKey));
    }

    @Override
    public String getId() {
      return getClass().getName();
    }
  }
}
