/*
 * 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