content: /*
   * 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.calcite.adapter.geode.util;
  
  import org.apache.calcite.avatica.util.DateTimeUtils;
  import org.apache.calcite.linq4j.tree.Primitive;
  import org.apache.calcite.rel.type.RelDataType;
  import org.apache.calcite.rel.type.RelDataTypeField;
  import org.apache.calcite.util.Util;
  
  import org.apache.commons.lang3.Strings;
  import org.apache.geode.cache.CacheClosedException;
  import org.apache.geode.cache.GemFireCache;
  import org.apache.geode.cache.Region;
  import org.apache.geode.cache.RegionExistsException;
  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.query.Struct;
  import org.apache.geode.pdx.PdxInstance;
  import org.apache.geode.pdx.ReflectionBasedAutoSerializer;
  
  import org.checkerframework.checker.nullness.qual.Nullable;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.lang.reflect.Field;
  import java.lang.reflect.Type;
  import java.util.Date;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Locale;
  import java.util.Map;
  import java.util.concurrent.ConcurrentHashMap;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * Utilities for the Geode adapter.
   */
  public class GeodeUtils {
  
    protected static final Logger LOGGER = LoggerFactory.getLogger(GeodeUtils.class.getName());
  
    /**
     * Cache for the client proxy regions created in the current ClientCache.
     */
    private static final Map<String, Region> REGION_MAP = new ConcurrentHashMap<>();
  
    private static String currentLocatorHost = "";
    private static int currentLocatorPort = -1;
  
    private static final JavaTypeFactoryExtImpl JAVA_TYPE_FACTORY = new JavaTypeFactoryExtImpl();
  
    private GeodeUtils() {
    }
  
    /**
     * Creates a Geode client instance connected to locator and configured to
     * support PDX instances.
     *
     * <p>If an old instance exists, it will be destroyed and re-created.
     *
     * @param locatorHost               Locator's host address
     * @param locatorPort               Locator's port
     * @param autoSerializerPackagePath package name of the Domain classes loaded in the regions
     * @return Returns a Geode {@link ClientCache} instance connected to Geode cluster
     */
    public static synchronized ClientCache createClientCache(String locatorHost,
        int locatorPort, String autoSerializerPackagePath,
        boolean readSerialized) {
      if (locatorPort != currentLocatorPort
          || !Strings.CI.equals(currentLocatorHost, locatorHost)) {
        LOGGER.info("Close existing ClientCache ["
            + currentLocatorHost + ":" + currentLocatorPort + "] for new Locator connection at: ["
            + locatorHost + ":" + locatorPort + "]");
        currentLocatorHost = locatorHost;
        currentLocatorPort = locatorPort;
        closeClientCache();
      }
  
      try {
        // If exists returns the existing client cache. This requires that the pre-created
        // client proxy regions can also be resolved from the regionMap
        return ClientCacheFactory.getAnyInstance();
      } catch (CacheClosedException cce) {
        // Do nothing if there is no existing instance
      }
  
      return new ClientCacheFactory()
          .addPoolLocator(locatorHost, locatorPort)
          .setPdxSerializer(new ReflectionBasedAutoSerializer(autoSerializerPackagePath))
          .setPdxReadSerialized(readSerialized)
          .create();
    }
  
    public static synchronized void closeClientCache() {
      try {
        ClientCacheFactory.getAnyInstance().close();
      } catch (CacheClosedException cce) {
        // Do nothing if there is no existing instance
      }
      REGION_MAP.clear();
    }
  
    /**
     * Obtains a proxy pointing to an existing Region on the server.
     *
     * @param cache {@link GemFireCache} instance to interact with the Geode server
     * @param regionName  Name of the region to create proxy for.
     * @return Returns a Region proxy to a remote (on the Server) regions.
     */
    public static synchronized Region createRegion(GemFireCache cache, String regionName) {
      requireNonNull(cache, "cache");
      requireNonNull(regionName, "regionName");
      Region region = REGION_MAP.get(regionName);
      if (region == null) {
        try {
          region = ((ClientCache) cache)
              .createClientRegionFactory(ClientRegionShortcut.PROXY)
              .create(regionName);
        } catch (IllegalStateException | RegionExistsException e) {
          // means this is a server cache (probably part of embedded testing
          // or clientCache is passed directly)
          region = cache.getRegion(regionName);
        }
  
        REGION_MAP.put(regionName, region);
      }
  
      return region;
    }
  
    /**
     * Converts a Geode object into a Row tuple.
     *
     * @param relDataTypeFields Table relation types
     * @param geodeResultObject Object value returned by Geode query
     * @return List of objects values corresponding to the relDataTypeFields
     */
    public static @Nullable Object convertToRowValues(
        List<RelDataTypeField> relDataTypeFields, Object geodeResultObject) {
  
      Object values;
  
      if (geodeResultObject instanceof Struct) {
        values = handleStructEntry(relDataTypeFields, geodeResultObject);
      } else if (geodeResultObject instanceof PdxInstance) {
        values = handlePdxInstanceEntry(relDataTypeFields, geodeResultObject);
      } else {
        values = handleJavaObjectEntry(relDataTypeFields, geodeResultObject);
      }
  
      return values;
    }
  
    private static Object handleStructEntry(
        List<RelDataTypeField> relDataTypeFields, Object obj) {
  
      Struct struct = (Struct) obj;
  
      Object[] values = new Object[relDataTypeFields.size()];
  
      int index = 0;
      for (RelDataTypeField relDataTypeField : relDataTypeFields) {
        Type javaType = JAVA_TYPE_FACTORY.getJavaClass(relDataTypeField.getType());
        Object rawValue;
        try {
          rawValue = struct.get(relDataTypeField.getName());
        } catch (IllegalArgumentException e) {
          rawValue = "<error>";
          System.err.println("Could find field : " + relDataTypeField.getName());
          e.printStackTrace();
        }
        values[index++] = convert(rawValue, (Class) javaType);
      }
  
      if (values.length == 1) {
        return values[0];
      }
  
      return values;
    }
  
    private static Object handlePdxInstanceEntry(
        List<RelDataTypeField> relDataTypeFields, Object obj) {
  
      PdxInstance pdxEntry = (PdxInstance) obj;
  
      Object[] values = new Object[relDataTypeFields.size()];
  
      int index = 0;
      for (RelDataTypeField relDataTypeField : relDataTypeFields) {
        Type javaType = JAVA_TYPE_FACTORY.getJavaClass(relDataTypeField.getType());
        Object rawValue = pdxEntry.getField(relDataTypeField.getName());
        values[index++] = convert(rawValue, (Class) javaType);
      }
  
      if (values.length == 1) {
        return values[0];
      }
  
      return values;
    }
  
    @SuppressWarnings("CatchAndPrintStackTrace")
    private static @Nullable Object handleJavaObjectEntry(
        List<RelDataTypeField> relDataTypeFields, Object obj) {
  
      Class<?> clazz = obj.getClass();
      if (relDataTypeFields.size() == 1) {
        try {
          Field javaField = clazz.getDeclaredField(relDataTypeFields.get(0).getName());
          javaField.setAccessible(true);
          return javaField.get(obj);
        } catch (Exception e) {
          e.printStackTrace();
        }
        return null;
      }
  
      Object[] values = new Object[relDataTypeFields.size()];
  
      int index = 0;
      for (RelDataTypeField relDataTypeField : relDataTypeFields) {
        try {
          Field javaField = clazz.getDeclaredField(relDataTypeField.getName());
          javaField.setAccessible(true);
          values[index++] = javaField.get(obj);
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
      return values;
    }
  
    @SuppressWarnings("JavaUtilDate")
    private static Object convert(Object o, Class clazz) {
      if (o == null) {
        return null;
      }
      Primitive primitive = Primitive.of(clazz);
      if (primitive != null) {
        clazz = primitive.boxClass;
      } else {
        primitive = Primitive.ofBox(clazz);
      }
      if (clazz == null) {
        return o.toString();
      }
      if (Map.class.isAssignableFrom(clazz)
              && o instanceof PdxInstance) {
        // This is in case of nested Objects!
        return Util.toString(
                ((PdxInstance) o).getFieldNames(), "PDX[", ",", "]");
      }
      if (clazz.isInstance(o)) {
        return o;
      }
      if (o instanceof Date && primitive != null) {
        o = ((Date) o).getTime() / DateTimeUtils.MILLIS_PER_DAY;
      }
      if (o instanceof Number && primitive != null) {
        return primitive.number((Number) o);
      }
      return o;
    }
  
    /**
     * Extract the first entity of each Regions and use it to build a table types.
     *
     * @param region existing region
     * @return derived data type.
     */
    public static RelDataType autodetectRelTypeFromRegion(Region<?, ?> region) {
      requireNonNull(region, "region");
  
      // try to detect type using value constraints (if they exists)
      final Class<?> constraint = region.getAttributes().getValueConstraint();
      if (constraint != null && !PdxInstance.class.isAssignableFrom(constraint)) {
        return new JavaTypeFactoryExtImpl().createStructType(constraint);
      }
  
      final Iterator<?> iter;
      if (region.getAttributes().getPoolName() == null) {
        // means current cache is server (not ClientCache)
        iter = region.keySet().iterator();
      } else {
        // for ClientCache
        iter = region.keySetOnServer().iterator();
      }
  
      if (!iter.hasNext()) {
        String message = String.format(Locale.ROOT, "Region %s is empty, can't "
            + "autodetect type(s)", region.getName());
        throw new IllegalStateException(message);
      }
  
      final Object entry = region.get(iter.next());
      return createRelDataType(entry);
    }
  
    // Create Relational Type by inferring a Geode entry or response instance.
    private static RelDataType createRelDataType(Object regionEntry) {
      JavaTypeFactoryExtImpl typeFactory = new JavaTypeFactoryExtImpl();
      if (regionEntry instanceof PdxInstance) {
        return typeFactory.createPdxType((PdxInstance) regionEntry);
      } else {
        return typeFactory.createStructType(regionEntry.getClass());
      }
    }
  
  }
  