text: /*
   * 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.
   */
  
  #define ROOT_NAME "testThinClientCacheables"
  #define ROOT_SCOPE DISTRIBUTED_ACK
  
  #include "fw_dunit.hpp"
  #include "BuiltinCacheableWrappers.hpp"
  #include "Utils.hpp"
  
  #include <ace/OS.h>
  #include <ace/High_Res_Timer.h>
  
  #include <string>
  
  #include "CacheHelper.hpp"
  
  using namespace apache::geode::client;
  using namespace test;
  
  #define CLIENT1 s1p1
  #define CLIENT2 s1p2
  #define SERVER1 s2p1
  
  CacheHelper* cacheHelper = nullptr;
  bool isLocalServer = false;
  
  #if defined(WIN32)
  // because we run out of memory on our pune windows desktops
  #define DEFAULTNUMKEYS 5
  #else
  #define DEFAULTNUMKEYS 15
  #endif
  #define KEYSIZE 256
  #define VALUESIZE 1024
  
  void initClient(const bool isthinClient) {
    if (cacheHelper == nullptr) {
      cacheHelper = new CacheHelper(isthinClient);
    }
    ASSERT(cacheHelper, "Failed to create a CacheHelper client instance.");
  }
  void cleanProc() {
    if (cacheHelper != nullptr) {
      delete cacheHelper;
      cacheHelper = nullptr;
    }
  }
  
  CacheHelper* getHelper() {
    ASSERT(cacheHelper != nullptr, "No cacheHelper initialized.");
    return cacheHelper;
  }
  
  void createRegion(const char* name, bool ackMode,
                    bool clientNotificationEnabled = false) {
    LOG("createRegion() entered.");
    fprintf(stdout, "Creating region --  %s  ackMode is %d\n", name, ackMode);
    fflush(stdout);
    auto regPtr = getHelper()->createRegion(name, ackMode, true, nullptr,
                                            clientNotificationEnabled);
    ASSERT(regPtr != nullptr, "Failed to create region.");
    LOG("Region created.");
  }
  
  void checkGets(int maxKeys, int8_t keyTypeId, int8_t valTypeId,
                 const std::shared_ptr<Region>& dataReg,
                 const std::shared_ptr<Region>& verifyReg) {
    for (int i = 0; i < maxKeys; i++) {
      CacheableWrapper* tmpkey =
          CacheableWrapperFactory::createInstance(keyTypeId);
      ASSERT(tmpkey != nullptr, "tmpkey is nullptr");
      CacheableWrapper* tmpval =
          CacheableWrapperFactory::createInstance(valTypeId);
      ASSERT(tmpval != nullptr, "tmpval is nullptr");
      tmpkey->initKey(i, KEYSIZE);
      auto key = std::dynamic_pointer_cast<CacheableKey>(tmpkey->getCacheable());
      auto val = dataReg->get(key);
      // also check that value is in local cache
      auto entry = dataReg->getEntry(key);
      ASSERT(entry != nullptr, "entry is nullptr");
      auto localVal = entry->getValue();
      uint32_t keychksum = tmpkey->getCheckSum();
      auto int32val = std::dynamic_pointer_cast<CacheableInt32>(
          verifyReg->get(static_cast<int32_t>(keychksum)));
      if (int32val == nullptr) {
        printf("GetsTask::keychksum: %u, key: %s\n", keychksum,
               Utils::nullSafeToString(key).c_str());
        FAIL("Could not find the checksum for the given key.");
      }
      uint32_t valchksum = static_cast<uint32_t>(int32val->value());
      uint32_t gotValChkSum = tmpval->getCheckSum(val);
      uint32_t gotLocalValChkSum = tmpval->getCheckSum(localVal);
      ASSERT(valchksum == gotValChkSum, "Expected valchksum == gotValChkSum");
      ASSERT(valchksum == gotLocalValChkSum,
             "Expected valchksum == gotLocalValChkSum");
      delete tmpkey;
      delete tmpval;
    }
  }
  
  const char* regionNames[] = {"DistRegionAck", "DistRegionNoAck"};
  
  const bool USE_ACK = true;
  const bool NO_ACK = false;
  
  DUNIT_TASK_DEFINITION(SERVER1, CreateServer1)
    {
      if (isLocalServer) CacheHelper::initServer(1);
      LOG("SERVER1 started");
    }
  END_TASK_DEFINITION
  
  DUNIT_TASK_DEFINITION(CLIENT1, StepOne)
    {
      initClient(true);
      createRegion(regionNames[0], USE_ACK);
      createRegion(regionNames[1], NO_ACK);
      CacheableHelper::registerBuiltins();
      LOG("StepOne complete.");
    }
  END_TASK_DEFINITION
  
  DUNIT_TASK_DEFINITION(CLIENT2, StepTwo)
    {
      initClient(true);
      createRegion(regionNames[0], USE_ACK);
      createRegion(regionNames[1], NO_ACK);
      CacheableHelper::registerBuiltins();
      LOG("StepTwo complete.");
    }
  END_TASK_DEFINITION
  
  DUNIT_TASK_DEFINITION(CLIENT1, PutsTask)
    {
      LOG("PutsTask started.");
      static int taskIndexPut = 0;
  
      auto keyTypes = CacheableWrapperFactory::getRegisteredKeyTypes();
      auto valueTypes = CacheableWrapperFactory::getRegisteredValueTypes();
  
      size_t keyTypeIndex = taskIndexPut / valueTypes.size();
      size_t valueTypeIndex = taskIndexPut % valueTypes.size();
  
      int8_t keyTypeId = keyTypes[keyTypeIndex];
      int8_t valTypeId = valueTypes[valueTypeIndex];
  
      printf("PutsTask::keyType = %s and valType = %s and taskIndexPut = %d\n",
             CacheableWrapperFactory::getTypeForId(keyTypeId).c_str(),
             CacheableWrapperFactory::getTypeForId(valTypeId).c_str(),
             taskIndexPut);
  
      CacheableWrapper* key = CacheableWrapperFactory::createInstance(keyTypeId);
      int maxKeys =
          (key->maxKeys() < DEFAULTNUMKEYS ? key->maxKeys() : DEFAULTNUMKEYS);
      delete key;
  
      auto dataReg = getHelper()->getRegion(regionNames[0]);
      auto verifyReg = getHelper()->getRegion(regionNames[1]);
      for (int i = 0; i < maxKeys; i++) {
        CacheableWrapper* tmpkey =
            CacheableWrapperFactory::createInstance(keyTypeId);
        CacheableWrapper* tmpval =
            CacheableWrapperFactory::createInstance(valTypeId);
        tmpkey->initKey(i, KEYSIZE);
        tmpval->initRandomValue(CacheableHelper::random(VALUESIZE) + 1);
        ASSERT(tmpkey->getCacheable() != nullptr,
               "tmpkey->getCacheable() is nullptr");
        // we can have nullptr values now after fix for bug #294
        if (tmpval->getCacheable() != nullptr) {
          dataReg->put(
              std::dynamic_pointer_cast<CacheableKey>(tmpkey->getCacheable()),
              tmpval->getCacheable());
        } else {
          try {
            dataReg->destroy(
                std::dynamic_pointer_cast<CacheableKey>(tmpkey->getCacheable()));
          } catch (const EntryNotFoundException&) {
            // expected
          }
          dataReg->create(
              std::dynamic_pointer_cast<CacheableKey>(tmpkey->getCacheable()),
              tmpval->getCacheable());
        }
        uint32_t keychksum = tmpkey->getCheckSum();
        uint32_t valchksum = tmpval->getCheckSum();
        verifyReg->put(static_cast<int32_t>(keychksum),
                       static_cast<int32_t>(valchksum));
        // also check that value is in local cache
        auto entry = dataReg->getEntry(
            std::dynamic_pointer_cast<CacheableKey>(tmpkey->getCacheable()));
        std::shared_ptr<Cacheable> localVal;
        if (entry != nullptr) {
          localVal = entry->getValue();
        }
        uint32_t localValChkSum = tmpval->getCheckSum(localVal);
        ASSERT(valchksum == localValChkSum,
               "Expected valchksum == localValChkSum");
        delete tmpkey;
        delete tmpval;
      }
      taskIndexPut++;
      LOG("PutsTask completed.");
    }
  END_TASK_DEFINITION
  
  DUNIT_TASK_DEFINITION(CLIENT2, GetsTask)
    {
      LOG("GetsTask started.");
      static int taskIndexGet = 0;
  
      auto keyTypes = CacheableWrapperFactory::getRegisteredKeyTypes();
      auto valueTypes = CacheableWrapperFactory::getRegisteredValueTypes();
  
      size_t keyTypeIndex = taskIndexGet / valueTypes.size();
      size_t valueTypeIndex = taskIndexGet % valueTypes.size();
  
      int8_t keyTypeId = keyTypes[keyTypeIndex];
      int8_t valTypeId = valueTypes[valueTypeIndex];
  
      printf("GetsTask::keyType = %s and valType = %s and taskIndexGet = %d\n",
             CacheableWrapperFactory::getTypeForId(keyTypeId).c_str(),
             CacheableWrapperFactory::getTypeForId(valTypeId).c_str(),
             taskIndexGet);
  
      CacheableWrapper* key = CacheableWrapperFactory::createInstance(keyTypeId);
      int maxKeys =
          (key->maxKeys() < DEFAULTNUMKEYS ? key->maxKeys() : DEFAULTNUMKEYS);
      delete key;
  
      auto dataReg = getHelper()->getRegion(regionNames[0]);
      auto verifyReg = getHelper()->getRegion(regionNames[1]);
      dataReg->localInvalidateRegion();
      verifyReg->localInvalidateRegion();
      checkGets(maxKeys, keyTypeId, valTypeId, dataReg, verifyReg);
  
      // Also check after running a region query. This ensures that the values
      // have deserialized on server so checks serialization/deserialization
      // compatibility with java server.
      std::string queryStr = "SELECT DISTINCT iter.key, iter.value FROM /";
      queryStr += ((std::string)regionNames[0] + ".entrySet AS iter");
      dataReg->query(queryStr.c_str());
      dataReg->localInvalidateRegion();
      verifyReg->localInvalidateRegion();
      checkGets(maxKeys, keyTypeId, valTypeId, dataReg, verifyReg);
  
      taskIndexGet++;
      LOG("GetsTask completed.");
    }
  END_TASK_DEFINITION
  
  DUNIT_TASK_DEFINITION(CLIENT1, CloseCache1)
    { cleanProc(); }
  END_TASK_DEFINITION
  
  DUNIT_TASK_DEFINITION(CLIENT2, CloseCache2)
    { cleanProc(); }
  END_TASK_DEFINITION
  
  DUNIT_TASK_DEFINITION(SERVER1, CloseServer1)
    {
      if (isLocalServer) {
        CacheHelper::closeServer(1);
        LOG("SERVER1 stopped");
      }
    }
  END_TASK_DEFINITION
  
  DUNIT_MAIN
    {
      CacheableHelper::registerBuiltins();
      CALL_TASK(CreateServer1);
      CALL_TASK(StepOne);
      CALL_TASK(StepTwo);
      size_t totKeyTypes =
          CacheableWrapperFactory::getRegisteredKeyTypes().size();
      size_t totValTypes =
          CacheableWrapperFactory::getRegisteredValueTypes().size();
      for (size_t i = 0; i < totKeyTypes; i++) {
        for (size_t j = 0; j < totValTypes; j++) {
          CALL_TASK(PutsTask);
          CALL_TASK(GetsTask);
        }
      }
      CALL_TASK(CloseCache1);
      CALL_TASK(CloseCache2);
      CALL_TASK(CloseServer1);
    }
  END_MAIN
  