text:
// ======================================================================
/*!
 * \brief Implementation of class Locus::Query
 */
// ======================================================================

#include "Query.h"
#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string/case_conv.hpp>
#include <boost/algorithm/string/classification.hpp>
#include <boost/algorithm/string/split.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/locale.hpp>
#include <boost/make_shared.hpp>
#include <macgyver/Exception.h>
#include <macgyver/StringConversion.h>
#include <algorithm>
#include <cmath>
#include <stdexcept>

using namespace std;

namespace
{
const char* CLIENT_ENCODING = "UTF8";

// ----------------------------------------------------------------------
/*!
 * \brief Convert from UTF-8 to given locale
 */
// ----------------------------------------------------------------------

string from_utf(const string& name, const string& ansiname, const string& encoding)
{
  try
  {
    try
    {
      return boost::locale::conv::from_utf(name, encoding, boost::locale::conv::stop);
    }
    catch (...)
    {
      return ansiname;
    }
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

string from_utf(const string& name, const string& encoding)
{
  try
  {
    return boost::locale::conv::from_utf(name, encoding);
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * \brief Utility algorithm to test if a container has a specific element
 */
// ----------------------------------------------------------------------

template <typename T, typename S>
bool contains(const T& theContainer, const S& theObject)
{
  try
  {
    const auto it = find(theContainer.begin(), theContainer.end(), theObject);
    return (it != theContainer.end());
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}
}  // namespace

namespace Locus
{
// ----------------------------------------------------------------------
/*!
 * \brief Default locale
 */
// ----------------------------------------------------------------------

boost::locale::generator locale_generator;
std::locale default_locale = locale_generator("fi_FI.UTF-8");

// ----------------------------------------------------------------------
/*!
 * \brief Default search radius
 */
// ----------------------------------------------------------------------

const float Query::default_radius = 50;

// ----------------------------------------------------------------------
/*!
 * \brief Alternate host constructor
 */
// ----------------------------------------------------------------------

Query::Query(const string& theHost,
             const string& theUser,
             const string& thePass,
             const string& theDatabase)
    : conn(new Fmi::Database::PostgreSQLConnection)
{
  try
  {
    /* Make a connection to the database */
    Fmi::Database::PostgreSQLConnectionOptions opt;
    opt.host = theHost;
    opt.username = theUser;
    opt.password = thePass;
    opt.database = theDatabase;
    opt.encoding = CLIENT_ENCODING;
    conn->open(opt);
    // conn.open(theHost, theUser, thePass, theDatabase, CLIENT_ENCODING);
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * \brief Alternate host constructor
 */
// ----------------------------------------------------------------------

Query::Query(const string& theHost,
             const string& theUser,
             const string& thePass,
             const string& theDatabase,
             const string& thePort)
    : conn(new Fmi::Database::PostgreSQLConnection)
{
  try
  {
    /* Make a connection to the database */
    Fmi::Database::PostgreSQLConnectionOptions opt;
    opt.host = theHost;
    opt.port = boost::lexical_cast<unsigned int>(thePort);
    opt.username = theUser;
    opt.password = thePass;
    opt.database = theDatabase;
    opt.encoding = CLIENT_ENCODING;
    conn->open(opt);
    // conn.open(theHost, theUser, thePass, theDatabase, CLIENT_ENCODING, thePort);
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * \brief Activate the correct language
 */
// ----------------------------------------------------------------------

void Query::SetOptions(const QueryOptions& theOptions)
{
  try
  {
    conn->setClientEncoding(theOptions.GetCharset());
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * SetDebug-mode on or off. Debug-mode prints sql-queries and
 * error-messages
 *
 * \param theFlag Boolean for mode: true=on, false=off
 */
// ----------------------------------------------------------------------

void Query::SetDebug(bool theFlag)
{
  try
  {
    debug = theFlag;
    conn->setDebug(theFlag);
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Resolves name for the feature code
 *
 * \param theCode Feature code
 * \return feature Name of the feature or false if not found
 */
// ----------------------------------------------------------------------

string Query::ResolveFeature(const QueryOptions& theOptions, const string& theCode)
{
  try
  {
    map<SQLQueryParameterId, boost::any> params;
    params[eQueryOptions] = theOptions;
    params[eFeatureCode] = theCode;

    string sqlStmt = constructSQLStatement(eResolveFeature, params);
    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    string retval;

    if (!res.empty())
      retval = res[0][0].as<string>();

    return retval;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Helper method to return variant name for search result using
 * correct language
 *
 * \param theId Database id for geoname
 * \return feature Name of the feature or false if not found
 */
// ----------------------------------------------------------------------

string Query::ResolveNameVariant(const QueryOptions& theOptions,
                                 const string& theId,
                                 const string& theSearchWord)
{
  try
  {
    // Uses name length to select shortest variant for language,
    // because there may be multiple variants like Tokio and
    // Tokion prefektuuri.

    map<SQLQueryParameterId, boost::any> params;
    params[eQueryOptions] = theOptions;
    params[eGeonamesId] = theId;
    params[eSearchWord] = theSearchWord;

    string sqlStmt = constructSQLStatement(eResolveNameVariant, params);
    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    string retval;

    if (!res.empty())
      retval = res[0][0].as<string>();

    return retval;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Helper method to return fmisid for a search result
 *
 * \param theId Database id for geoname
 * \return fmisid optional fmisid
 */
// ----------------------------------------------------------------------

boost::optional<int> Query::ResolveFmisid(const QueryOptions& theOptions, const string& theId)
{
  try
  {
    map<SQLQueryParameterId, boost::any> params;
    params[eQueryOptions] = theOptions;
    params[eGeonamesId] = theId;
    string sqlStmt = constructSQLStatement(eResolveFmisid, params);

    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    if (res.empty())
      return {};

    return res[0][0].as<int>();
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Resolves name for country
 *
 * \param theIsoCode Iso country code
 * \return country Name of the country or false if not found
 */
// ----------------------------------------------------------------------

string Query::ResolveCountry(const QueryOptions& theOptions, const string& theIsoCode)
{
  try
  {
    map<SQLQueryParameterId, boost::any> params;
    params[eQueryOptions] = theOptions;
    params[eCountryIso2Code] = theIsoCode;

    string sqlStmt = constructSQLStatement(eResolveCountry1, params);
    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    string retval;

    if (!res.empty())
    {
      retval = res[0][0].as<string>();
    }
    else
    {
      // If variant is not found use name in countries-table
      string localsqlStmt = constructSQLStatement(eResolveCountry2, params);
      pqxx::result localres = conn->executeNonTransaction(localsqlStmt);

      if (!localres.empty())
        retval = localres[0][0].as<string>();
    }

    return retval;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Helper method that resolves name for municipality id
 * \param theId Municipality id
 * \return municipality Name of the municipality or "" if not found
 */
// ----------------------------------------------------------------------

string Query::ResolveMunicipality(const QueryOptions& theOptions, const string& theId)
{
  try
  {
    map<SQLQueryParameterId, boost::any> params;
    params[eQueryOptions] = theOptions;
    params[eMunicipalityId] = theId;

    string sqlStmt = constructSQLStatement(eResolveMunicipality1, params);
    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    string retval;

    if (!res.empty())
    {
      retval = res[0][0].as<string>();
    }

    if (theOptions.GetLanguage() == "fi")
      return retval;

    sqlStmt = constructSQLStatement(eResolveMunicipality2, params);
    res = conn->executeNonTransaction(sqlStmt);

    // default is the original name
    if (!res.empty())
      retval = res[0][0].as<string>();

    return retval;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Resolves name for administrative area (state)
 *
 * \param theCode Administrative area code
 * \param theCountry Country iso code
 * \return administrative Name of the state or false if not found
 */
// ----------------------------------------------------------------------

string Query::ResolveAdministrative(const string& theCode, const string& theCountry)
{
  try
  {
    map<SQLQueryParameterId, boost::any> params;
    params[eQueryOptions] = QueryOptions();
    string adminCode = theCountry + "." + theCode;
    params[eAdminCode] = adminCode;

    string sqlStmt = constructSQLStatement(eResolveAdministrative, params);
    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    string retval;

    if (!res.empty())
    {
      retval = res[0][0].as<string>();
    }

    return retval;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Helper method that generates country conditions for sql-query
 *
 * \param theQuery the SQL query yo modify
 */
// ----------------------------------------------------------------------

void Query::AddCountryConditions(const QueryOptions& theOptions, string& theQuery) const
{
  try
  {
    // Nothing added if % or all is in the list
    const list<string>& countries = theOptions.GetCountries();
    if (!countries.empty() && !contains(countries, "%") && !contains(countries, "all"))
    {
      // Append to the query

      int n = 1;
      for (auto it = countries.begin(); it != countries.end(); ++it, ++n)
      {
        string country_iso2 = *it;
        Fmi::ascii_toupper(country_iso2);

        if (n == 1)
          theQuery += " AND geonames.countries_iso2 IN (";
        else
          theQuery += ",";
        theQuery += conn->quote(country_iso2);
      }

      theQuery += ")";
    }

    const list<string>& excluded_countries = theOptions.GetExcludedCountries();
    if (!excluded_countries.empty() && !contains(countries, "%") && !contains(countries, "all"))
    {
      // Append to the query

      int n = 1;
      for (auto it = excluded_countries.begin(); it != excluded_countries.end(); ++it, ++n)
      {
        string country_iso2 = *it;
        Fmi::ascii_tolower(country_iso2);

        if (n == 1)
          theQuery += " AND geonames.countries_iso2 NOT IN (";
        else
          theQuery += ",";
        theQuery += conn->quote(country_iso2);
      }

      theQuery += ")";
    }
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Helper method that generates feature conditions for sql-query
 *
 */
// ----------------------------------------------------------------------
void Query::AddFeatureConditions(const QueryOptions& theOptions, string& theQuery) const
{
  try
  {
    // Nothing added if % or all is in the list

    const list<string>& features = theOptions.GetFeatures();

    if (features.empty() || contains(features, "%") || contains(features, "all"))
      return;

    // Append to the query

    int n = 1;
    for (auto it = features.begin(); it != features.end(); ++it, ++n)
    {
      theQuery += (n == 1 ? " AND (" : " OR ");
      theQuery += "features_code=";
      theQuery += conn->quote(*it);
    }

    theQuery += ")";
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Helper method that generates keyword conditions for sql-query
 *
 * \param theQuery sql-string
 */
// ----------------------------------------------------------------------

void Query::AddKeywordConditions(const QueryOptions& theOptions, string& theQuery) const
{
  try
  {
    const list<string>& keywords = theOptions.GetKeywords();

    if (keywords.empty() || contains(keywords, "%") || contains(keywords, "all"))
      return;

    // Append to the query

    int n = 1;
    for (const auto& keyword : keywords)
    {
      if (n == 1)
      {
        theQuery +=
            " AND geonames.id IN (SELECT geonames_id FROM keywords_has_geonames WHERE keyword IN (";
        theQuery += conn->quote(keyword);
      }
      else
      {
        theQuery += ",";
        theQuery += conn->quote(keyword);
      }
      n++;
    }

    if (n > 1)
      theQuery += "))";
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Public method for fetching locations by name
 *
 * \param theName Place name
 * \return locations Array of SimpleLocation objects
 */
// ----------------------------------------------------------------------

Query::return_type Query::FetchByName(const QueryOptions& theOptions, const string& theName)
{
  try
  {
    map<SQLQueryParameterId, boost::any> params;
    QueryOptions opts = theOptions;
    if (!opts.GetNameType().empty())
      opts.SetLanguage(opts.GetNameType());
    params[eQueryOptions] = opts;
    params[eLocationName] = theName;

    SetOptions(theOptions);

    // This allows queries like Helsinki, Finland

    vector<string> qparts;
    if (!theName.empty())
      boost::algorithm::split(qparts, theName, boost::algorithm::is_any_of(","));
    string searchword = (qparts.empty() ? string("") : qparts[0]);

    params[eSearchWord] = searchword;

    string collation = theOptions.GetCollation();

    // Set country priorities

    const list<string>& countries = theOptions.GetCountries();

    std::string country_priorities;
    if (countries.size() > 1)
    {
      country_priorities += ", CASE countries_iso2 WHEN ";
      int n = 1;
      for (auto it = countries.begin(); it != countries.end(); ++it, ++n)
      {
        if (n == 1)
        {
          country_priorities += conn->quote(*it);
          country_priorities += " THEN 1 ";
        }
        else
        {
          country_priorities += " WHEN ";
          country_priorities += conn->quote(*it);
          country_priorities += " THEN ";
          country_priorities += Fmi::to_string(n);
        }
      }
      if (n > 1)
        country_priorities += " ELSE 1000 END as country_priority ";
    }
    params[eCountryPriorities] = country_priorities;

    // Set feature priorities

    const list<string>& features = theOptions.GetFeatures();

    std::string feature_priorities;
    if (features.size() > 1)
    {
      feature_priorities += ", CASE features_code WHEN ";
      int n = 1;
      for (auto it = features.begin(); it != features.end(); ++it, ++n)
      {
        if (n == 1)
        {
          feature_priorities += conn->quote(*it);
          feature_priorities += " THEN 1 ";
        }
        else
        {
          feature_priorities += " WHEN ";
          feature_priorities += conn->quote(*it);
          feature_priorities += " THEN ";
          feature_priorities += Fmi::to_string(n);
        }
      }
      if (n > 1)
        feature_priorities += " ELSE 1000 END as feature_priority ";
    }
    params[eFeaturePriorities] = feature_priorities;

    string sqlStmt = constructSQLStatement(eFetchByName, params);

    // std::cout << "SQL:\n" << sqlStmt << "\n\n\n";

    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    // Create result list
    return_type locations;
    if (qparts.size() == 2)
      locations = build_locations(theOptions, res, searchword, qparts[1]);
    else
      locations = build_locations(theOptions, res, searchword);

    if (!locations.empty())
      return locations;

    // Prevent endless recursion

    if (recursive_query)
      return locations;

    if (theOptions.GetFullCountrySearch())
    {
      // Search all countries

      if (debug)
        cout << "Do full country seach because limited search didn't return results" << endl;

      QueryOptions newoptions = theOptions;
      newoptions.SetCountries("%");
      recursive_query = true;
      locations = FetchByName(newoptions, theName);
      recursive_query = false;
    }

    return locations;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Alias for FetchByLonLat-method because some people don't like lon,lat order
 *
 * \param theLatitude Latitude
 * \param theLongitude Longitude
 * \param theRadius Maximum distance from point in kilometers.
 * \return locations Array of SimpleLocation objects
 */
// ----------------------------------------------------------------------

Query::return_type Query::FetchByLatLon(const QueryOptions& theOptions,
                                        float theLatitude,
                                        float theLongitude,
                                        float theRadius)
{
  try
  {
    return FetchByLonLat(theOptions, theLongitude, theLatitude, theRadius);
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Method for fetching locations close to some lon,lat point
 *
 * \param theLatitude Latitude
 * \param theLongitude Longitude
 * \param theRadius Maximum distance from point in kilometers.
 * \return locations Array of SimpleLocation objects
 */
// ----------------------------------------------------------------------

Query::return_type Query::FetchByLonLat(const QueryOptions& theOptions,
                                        float theLongitude,
                                        float theLatitude,
                                        float theRadius)
{
  try
  {
    map<SQLQueryParameterId, boost::any> params;
    params[eQueryOptions] = theOptions;
    params[eLongitude] = theLongitude;
    params[eLatitude] = theLatitude;
    params[eRadius] = theRadius;

    SetOptions(theOptions);

    string sqlStmt = constructSQLStatement(eFetchByLonLat, params);
    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    auto ret = build_locations(theOptions, res, "", "");

    auto copyret = boost::make_shared<Locus::Query::return_type>(ret);

    return ret;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Method for fetching location by unique id
 *
 * \param theId unique fminames id for location
 * \return locations Array of SimpleLocation objects (0/1)
 */
// ----------------------------------------------------------------------

Query::return_type Query::FetchById(const QueryOptions& theOptions, int theId)
{
  try
  {
    SetOptions(theOptions);

    map<SQLQueryParameterId, boost::any> params;
    params[eQueryOptions] = theOptions;
    params[eGeonameId] = theId;

    string sqlStmt = constructSQLStatement(eFetchById, params);
    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    if (res.empty() && theId >= 10000000)
      return FetchById(theOptions, -theId);

    return build_locations(theOptions, res, "", "");
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Method for fetching locations by keyword
 *
 * \param theKeyword keyword
 * \return locations Array of SimpleLocation objects
 */
// ----------------------------------------------------------------------

Query::return_type Query::FetchByKeyword(const QueryOptions& theOptions, const string& theKeyword)
{
  try
  {
    SetOptions(theOptions);

    // We always want all the names in the keyword, not just the default 100
    QueryOptions options = theOptions;
    options.SetResultLimit(0);

    map<SQLQueryParameterId, boost::any> params;
    params[eKeyword] = theKeyword;
    params[eQueryOptions] = options;

    auto locations = boost::make_shared<Locus::Query::return_type>();

    string sqlStmt = constructSQLStatement(eFetchByKeyword1, params);
    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    if (res.size() != 1)
      return *locations;

    sqlStmt = constructSQLStatement(eFetchByKeyword2, params);

    res = conn->executeNonTransaction(sqlStmt);

    auto ret = build_locations(options, res, "", "");

    return ret;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * Method for fetching number of locations by keyword
 *
 * \param theKeyword keyword
 * \return locations Array of SimpleLocation objects
 */
// ----------------------------------------------------------------------

unsigned int Query::CountKeywordLocations(const QueryOptions& theOptions, const string& theKeyword)
{
  try
  {
    SetOptions(theOptions);

    map<SQLQueryParameterId, boost::any> params;
    params[eQueryOptions] = theOptions;
    params[eKeyword] = theKeyword;

    auto locations = boost::make_shared<Locus::Query::return_type>();

    string sqlStmt = constructSQLStatement(eCountKeywordLocations, params);
    pqxx::result res = conn->executeNonTransaction(sqlStmt);

    return res[0]["count"].as<unsigned int>();
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

// ----------------------------------------------------------------------
/*!
 * \brief Build a list of locations from query result
 *
 * Note: This implementation differs from the PHP version in that
 *       we implement a cache for storing query results which are
 *       likely to be repeated for example in keyword searches.
 */
// ----------------------------------------------------------------------

Query::return_type Query::build_locations(const QueryOptions& theOptions,
                                          const pqxx::result& theR,
                                          const string& theSearchWord,
                                          const string& theArea /* = ""*/)
{
  try
  {
    return_type locations;

    if (theR.empty())
      return locations;

    // Does the result have a field for overriding names?
    bool has_override_field = false;
    for (unsigned int field = 0; !has_override_field && field < theR.columns(); ++field)
    {
      if (string(theR.column_name(field)) == string("override_name"))
        has_override_field = true;
    }

    // Caches subquery results

    map<string, string> country_cache;       // iso2 --> country
    map<string, string> feature_cache;       // feature_code --> description
    map<string, string> municipality_cache;  // id --> municipality
    map<string, string> admin_cache;         // name|iso2 --> admin

    // Process one location at a time

    for (pqxx::result::const_iterator row = theR.begin(); row != theR.end(); ++row)
    {
      // Do not handle locations without timezones. This is just a safety check,
      // NULL timezones should be removed already in the SQL query, otherwise
      // you might get zero results if the result count limit is 1.

      if (row["timezone"].is_null())
        continue;

      // Determine name
      string name = (!row["name"].is_null() ? row["name"].as<string>() : "NULL");

      // Use override if there is one

      bool override_done = false;
      if (has_override_field)
      {
        string altname =
            (!row["override_name"].is_null() ? row["override_name"].as<string>() : "NULL");
        if (!altname.empty() && altname != "NULL")
        {
          name = altname;
          override_done = true;
        }
      }

      // Search for possible translations

      if (!override_done && !theOptions.GetLanguage().empty())
      {
        string variant;
        auto id = row["id"].as<string>();
        if (!theOptions.GetAutoCompleteMode())
          variant = ResolveNameVariant(theOptions, id);
        else
          variant = ResolveNameVariant(theOptions, id, theSearchWord);

        if (!variant.empty())
          name = variant;
      }

      if ((!row["ansiname"].is_null()) && (theOptions.GetCharset() != "utf8"))
        name = from_utf(name, row["ansiname"].as<string>(), theOptions.GetCharset());

      // Elevation

      int elevation = 0;

      if (!row["elevation"].is_null() && (row["elevation"].as<int>() != 0))
        elevation = row["elevation"].as<int>();
      else if (!row["dem"].is_null())
        elevation = row["dem"].as<int>();

      // Country and description

      string country;
      string iso2;
      if (!row["iso2"].is_null())
      {
        iso2 = row["iso2"].as<string>();
        const auto pos = country_cache.find(iso2);
        if (pos != country_cache.end())
          country = pos->second;
        else
        {
          country = ResolveCountry(theOptions, iso2);
          if (theOptions.GetCharset() != "utf8")
            country = from_utf(country, theOptions.GetCharset());
          country_cache[iso2] = country;
        }
      }

      string description;
      string features_code;
      if (!row["features_code"].is_null())
      {
        features_code = row["features_code"].as<string>();
        const auto pos = feature_cache.find(features_code);
        if (pos != feature_cache.end())
          description = pos->second;
        else
        {
          description = ResolveFeature(theOptions, features_code);
          feature_cache[features_code] = description;
        }
      }

      // Administrative areas

      string administrative;
      if (!row["municipalities_id"].is_null())
      {
        auto municipalities_id = row["municipalities_id"].as<string>();

        if (!municipalities_id.empty())
        {
          const auto pos = municipality_cache.find(municipalities_id);
          if (pos != municipality_cache.end())
            administrative = pos->second;
          else
          {
            administrative = ResolveMunicipality(theOptions, municipalities_id);
            if (theOptions.GetCharset() != "utf8")
              administrative = from_utf(administrative, theOptions.GetCharset());
            municipality_cache[municipalities_id] = administrative;
          }
        }
        else
        {
          auto admin1 = row["admin1"].as<string>();
          if (!admin1.empty())
          {
            auto localiso2 = row["iso2"].as<string>();
            string key = admin1 + '|' + localiso2;
            const auto pos = admin_cache.find(key);
            if (pos != admin_cache.end())
              administrative = pos->second;
            else
            {
              administrative = ResolveAdministrative(admin1, localiso2);
              if (theOptions.GetCharset() != "utf8")
                administrative = from_utf(administrative, theOptions.GetCharset());
              admin_cache[key] = administrative;
            }
          }
        }
      }

      // Check if area is correct
      bool ok = true;

      if (!theArea.empty())
      {
        string lc_area = boost::locale::to_lower(theArea, default_locale);
        ok = (lc_area == boost::locale::to_lower(country, default_locale) ||
              lc_area == boost::locale::to_lower(administrative, default_locale));
      }

      if (ok)
      {
        SimpleLocation loc(name,
                           row["lon"].as<float>(),
                           row["lat"].as<float>(),
                           country,
                           features_code,
                           description,
                           row["timezone"].as<string>(),
                           administrative,
                           row["population"].as<unsigned int>(),
                           iso2,
                           row["id"].as<int>(),
                           elevation);

        loc.fmisid = ResolveFmisid(theOptions, row["id"].as<string>());

        locations.emplace_back(loc);
      }

      // See if locations-sequence is already long enough

      if (theOptions.GetResultLimit() > 0 && locations.size() >= theOptions.GetResultLimit())
      {
        break;
      }
    }

    // Sort exact matchs first if autocompletemode

    if (theOptions.GetAutoCompleteMode())
    {
      vector<SimpleLocation> bestmatches;
      vector<SimpleLocation> secondarymatches;

      // Remove "%" from searchword
      string tmp = theSearchWord.substr(0, theSearchWord.size() - 1);

      for (const auto& location : locations)
      {
        if (boost::iequals(tmp, location.name))
          bestmatches.push_back(location);
        else
          secondarymatches.push_back(location);
      }

      locations.clear();

      for (const auto& match : bestmatches)
        locations.push_back(match);

      for (const auto& match : secondarymatches)
        locations.push_back(match);
    }

    return locations;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

string Query::constructSQLStatement(SQLQueryId theQueryId,
                                    const map<SQLQueryParameterId, boost::any>& theParams)
{
  try
  {
    std::string sql;

    const auto& theOptions = boost::any_cast<const QueryOptions&>(theParams.at(eQueryOptions));

    switch (theQueryId)
    {
      case eResolveFeature:
      {
        auto theCode = boost::any_cast<string>(theParams.at(eFeatureCode));
        sql += "SELECT shortdesc FROM features WHERE code=";
        sql += conn->quote(theCode);
        break;
      }
      case eResolveNameVariant:
      {
        auto theGeonamesId = boost::any_cast<string>(theParams.at(eGeonamesId));
        auto theSearchWord = boost::any_cast<string>(theParams.at(eSearchWord));
        string language = theOptions.GetLanguage();
        Fmi::ascii_tolower(language);

        if (!theOptions.GetAutoCompleteMode())
        {
          sql +=
              "SELECT name,length(name) AS l, priority FROM alternate_geonames WHERE geonames_id=";
          sql += conn->quote(theGeonamesId);
          sql += " AND language=";
          sql += conn->quote(language);
          sql +=
              " AND historic=false ORDER BY priority ASC, preferred DESC, l ASC, name ASC LIMIT 1";
        }
        else
        {
          sql +=
              "SELECT name,length(name) As l, priority FROM alternate_geonames WHERE geonames_id=";
          sql += conn->quote(theGeonamesId);
          sql += " AND language=";
          sql += conn->quote(language);
          sql += " AND name LIKE ";
          sql += conn->quote(theSearchWord);
          sql +=
              " AND historic=false ORDER BY priority ASC, preferred DESC, l ASC, name ASC LIMIT 1";
        }
        break;
      }
      case eResolveFmisid:
      {
        auto theGeonamesId = boost::any_cast<string>(theParams.at(eGeonamesId));
        sql += "SELECT name FROM alternate_geonames WHERE language='fmisid' AND geonames_id=";
        sql += conn->quote(theGeonamesId);
        break;
      }
      case eResolveCountry1:
      case eResolveCountry2:
      {
        auto iso2 = boost::any_cast<string>(theParams.at(eCountryIso2Code));
        string language = theOptions.GetLanguage();
        Fmi::ascii_tolower(language);

        if (theQueryId == eResolveCountry1)
        {
          sql +=
              "SELECT alternate_geonames.name AS name, length(alternate_geonames.name) AS l"
              " FROM geonames, alternate_geonames"
              " WHERE geonames.features_code='PCLI'"
              " AND geonames.countries_iso2=";
          sql += conn->quote(iso2);
          sql +=
              " AND geonames.id=alternate_geonames.geonames_id"
              " AND alternate_geonames.language=";
          sql += conn->quote(language);
          sql += " ORDER BY preferred DESC, alternate_geonames.priority ASC, l ASC LIMIT 1";
        }
        else
        {
          sql += "SELECT name FROM countries WHERE iso2=";
          sql += conn->quote(iso2);
        }
        break;
      }
      case eResolveMunicipality1:
      case eResolveMunicipality2:
      {
        auto theMunicipalityId = boost::any_cast<string>(theParams.at(eMunicipalityId));
        string language = theOptions.GetLanguage();
        Fmi::ascii_tolower(language);

        if (theQueryId == eResolveMunicipality1)
        {
          sql += "SELECT name FROM municipalities WHERE id=";
          sql += conn->quote(theMunicipalityId);
        }
        else
        {
          sql += "SELECT name FROM alternate_municipalities WHERE municipalities_id=";
          sql += conn->quote(theMunicipalityId);
          sql += " AND language=";
          sql += conn->quote(language);
        }
        break;
      }
      case eResolveAdministrative:
      {
        auto theAdminCode = boost::any_cast<string>(theParams.at(eAdminCode));
        Fmi::ascii_tolower(theAdminCode);
        sql += "SELECT name FROM admin1codes WHERE code=";
        sql += conn->quote(theAdminCode);
        break;
      }
      case eFetchByName:
      {
        if (theOptions.GetSearchVariants())
          sql += "(";

        auto theSearchWord = boost::any_cast<string>(theParams.at(eSearchWord));
        auto theCountryPriorities = boost::any_cast<string>(theParams.at(eCountryPriorities));
        auto theFeaturePriorities = boost::any_cast<string>(theParams.at(eFeaturePriorities));

        sql +=
            "SELECT DISTINCT geonames.name AS name,"
            " geonames.ansiname AS ansiname,"
            " lat, lon, countries_iso2 AS iso2,"
            " features_code, timezone, geonames.id as id, geonames.priority as geonames_priority,"
            " municipalities_id, admin1, population, "
            " elevation, dem, "
            " CASE WHEN population>50000 THEN population ELSE 0 END AS population_priority ";
        sql += theCountryPriorities;
        sql += ' ';
        sql += theFeaturePriorities;
        sql += " FROM geonames WHERE LOWER(geonames.name) LIKE LOWER(";
        sql += conn->quote(theSearchWord);
        sql += ')';

        if (conn->collateSupported())
        {
          sql += " COLLATE ";
          sql += conn->quote(theOptions.GetCollation());
        }

        // PHP version does not do this, but we cannot tolerate it in brainstorm
        sql += " AND timezone IS NOT NULL";

        if (theOptions.GetPopulationMin() > 0)
        {
          sql += " AND population>=";
          sql += Fmi::to_string(theOptions.GetPopulationMin());
        }
        if (theOptions.GetPopulationMax() > 0)
        {
          sql += " AND population<=";
          sql += Fmi::to_string(theOptions.GetPopulationMax());
        }

        AddFeatureConditions(theOptions, sql);
        AddCountryConditions(theOptions, sql);
        AddKeywordConditions(theOptions, sql);

        if (theOptions.GetSearchVariants())
        {
          string language = theOptions.GetLanguage();
          Fmi::ascii_tolower(language);

          sql +=
              ") UNION (SELECT DISTINCT geonames.name AS name,"
              " geonames.ansiname AS ansiname, lat, lon,"
              " countries_iso2 AS iso2, features_code, timezone,"
              " geonames.id as id, geonames.priority as geonames_priority, municipalities_id,"
              " admin1, population, "
              " elevation, dem, "
              " CASE WHEN population>50000 THEN population ELSE 0 END AS population_priority ";
          sql += theCountryPriorities;
          sql += ' ';
          sql += theFeaturePriorities;
          sql +=
              " FROM geonames, alternate_geonames WHERE LOWER(alternate_geonames.name) LIKE LOWER(";
          sql += conn->quote(theSearchWord);
          sql += ")";

          if (conn->collateSupported())
          {
            sql += " COLLATE ";
            sql += conn->quote(theOptions.GetCollation());
          }

          sql +=
              " AND alternate_geonames.geonames_id=geonames.id AND alternate_geonames.language "
              "LIKE ";
          sql += conn->quote(language);

          if (theOptions.GetPopulationMin() > 0)
          {
            sql += " AND population>=";
            sql += Fmi::to_string(theOptions.GetPopulationMin());
          }
          if (theOptions.GetPopulationMax() > 0)
          {
            sql += " AND population<=";
            sql += Fmi::to_string(theOptions.GetPopulationMax());
          }
          if (theOptions.GetAutoCompleteMode())
          {
            sql += " AND alternate_geonames.language=";
            sql += conn->quote(language);
          }

          AddFeatureConditions(theOptions, sql);
          AddCountryConditions(theOptions, sql);
          AddKeywordConditions(theOptions, sql);
          sql += ')';
        }

        sql += " ORDER BY geonames_priority, population_priority DESC, ";
        sql += (theCountryPriorities.empty() ? "" : "country_priority, ");
        sql += (theFeaturePriorities.empty() ? "" : "feature_priority, ");
        sql += " population DESC, name";

        if (conn->collateSupported())
        {
          sql += " COLLATE ";
          sql += conn->quote(theOptions.GetCollation());
        }

        break;
      }
      case eFetchByLonLat:
      {
        auto theLongitude = boost::any_cast<float>(theParams.at(eLongitude));
        auto theLatitude = boost::any_cast<float>(theParams.at(eLatitude));
        auto theRadius = boost::any_cast<float>(theParams.at(eRadius));

        sql +=
            "WITH candidates AS ("
            "SELECT geonames.id AS id, geonames.name AS name,"
            "geonames.ansiname AS ansiname, lat, lon,"
            "countries_iso2 AS iso2, features_code, timezone,"
            "population, elevation, dem, municipalities_id,"
            "admin1, ST_Distance(ST_GeographyFromText('POINT(";
        sql += Fmi::to_string(theLongitude);
        sql += ' ';
        sql += Fmi::to_string(theLatitude);
        sql += ")'), the_geog) as distance FROM geonames WHERE ";

        // PHP version does not do this, but we cannot tolerate it in brainstorm
        sql += " timezone IS NOT NULL";

        if (theOptions.GetPopulationMin() > 0)
        {
          sql += " AND population>=";
          sql += Fmi::to_string(theOptions.GetPopulationMin());
        }
        if (theOptions.GetPopulationMax() > 0)
        {
          sql += " AND population<=";
          sql += Fmi::to_string(theOptions.GetPopulationMax());
        }

        AddCountryConditions(theOptions, sql);
        AddFeatureConditions(theOptions, sql);
        AddKeywordConditions(theOptions, sql);

        sql += " ORDER BY the_geom <-> ST_GeomFromText('POINT(";
        sql += Fmi::to_string(theLongitude);
        sql += ' ';
        sql += Fmi::to_string(theLatitude);
        sql += ")',4326)";

        // <-> ordering is appriximate, so we need to fetch more results
        // than wanted to make sure the final list is properly ordered.
        // HOWEVER: It is important to keep this number small, increasing
        // the margin increases execution time rapidly

        const int limit_safety_margin = 10;

        if (theOptions.GetResultLimit() > 0)
        {
          sql += " LIMIT ";
          sql += Fmi::to_string(limit_safety_margin + theOptions.GetResultLimit());
        }

        sql += ") SELECT * from candidates";

        // This best best done in the outer select since postgresql 9.1
        if (theRadius > 0)
        {
          sql += " WHERE distance<=";
          sql += Fmi::to_string(theRadius * 1000);
        }

        sql += " ORDER BY distance";
        if (theOptions.GetResultLimit() > 0)
        {
          sql += " LIMIT ";
          sql += Fmi::to_string(theOptions.GetResultLimit());
        }

        break;
      }
      case eFetchById:
      {
        auto theId = boost::any_cast<int>(theParams.at(eGeonameId));

        sql +=
            "SELECT id, name, ansiname, lat, lon, countries_iso2 AS iso2, features_code,"
            " timezone, municipalities_id, admin1, population,"
            " elevation, dem "
            " FROM geonames WHERE id=";
        sql += Fmi::to_string(theId);
        break;
      }
      case eFetchByKeyword1:
      case eFetchByKeyword2:
      case eFetchByKeyword3:
      {
        auto theKeyword = boost::any_cast<string>(theParams.at(eKeyword));

        if (theQueryId == eFetchByKeyword1)
        {
          sql += "SELECT keyword FROM keywords WHERE keyword=";
          sql += conn->quote(theKeyword);
        }
        else if (theQueryId == eFetchByKeyword2)
        {
          sql +=
              "SELECT geonames.id AS id, geonames.name AS name,\n "
              "geonames.ansiname AS ansiname,lat,lon,\n "
              "countries_iso2 AS iso2, features_code, timezone,\n "
              "population, elevation, dem,\n "
              "municipalities_id, admin1,\n "
              "keywords_has_geonames.name AS override_name FROM \n "
              "geonames, keywords_has_geonames WHERE\n "
              "keywords_has_geonames.keyword=";
          sql += conn->quote(theKeyword);
          sql +=
              " AND geonames.id=keywords_has_geonames.geonames_id"
              " ORDER BY name";
        }
        else if (theQueryId == eFetchByKeyword3)
        {
          // long version
          sql +=
              "SELECT georesults.*,\n"
              "       municipalities.name AS mname,\n"
              "       altname_translations.name AS altname,\n"
              "       alternate_municipalities.name AS altmname,\n"
              "       admin1codes.name AS adminname,\n"
              "       iso2_translations.name AS altcname\n"
              "\n"
              "FROM\n"
              "(\n"

              // -- basic geonames results

              "  SELECT geonames.admin1,\n"
              "         geonames.ansiname AS ansiname,\n"
              "         geonames.countries_iso2 AS iso2,\n"
              "         geonames.elevation,\n"
              "         geonames.features_code,\n"
              "         geonames.dem,\n"
              "         geonames.id AS id,\n"
              "         geonames.lat,\n"
              "         geonames.lon,\n"
              "         geonames.municipalities_id,\n"
              "         geonames.name AS name,\n"
              "         geonames.population,\n"
              "         geonames.timezone,\n"
              "         countries.name AS cname,\n"
              "         features.shortdesc AS shortdesc,\n"
              "         keywords_has_geonames.name AS override_name\n"
              "  FROM geonames, keywords_has_geonames, features, countries\n"
              "  WHERE geonames.id=keywords_has_geonames.geonames_id\n"
              "  AND keywords_has_geonames.keyword=";
          sql += conn->quote(theKeyword);
          sql +=
              "  AND features.code=geonames.features_code\n"
              "  AND geonames.countries_iso2=countries.iso2\n"
              ")\n"
              "AS georesults\n"

              // -- left join to add municipality if available

              "LEFT JOIN municipalities\n"
              "ON (municipalities.id=georesults.municipalities_id\n"
              "    AND municipalities.countries_iso2=georesults.iso2)\n"
              "\n"
              "-- left join to add alternate municipality if available\n"
              "\n"
              "LEFT JOIN alternate_municipalities\n"
              "ON (georesults.municipalities_id=alternate_municipalities.municipalities_id\n"
              "    AND alternate_municipalities.language=";
          sql += conn->quote(theOptions.GetLanguage());
          sql +=
              "   )\n"
              "\n"
              "-- left join to add admin name if available\n"
              "\n"
              "LEFT JOIN admin1codes\n"
              "ON (admin1codes.code=georesults.admin1 AND admin1codes.geonames_id=georesults.id)\n"

              //  left join to add alternate name if available

              "LEFT JOIN\n"
              "(\n"
              "  SELECT id,name FROM\n"
              "  (\n"
              "    SELECT geonames.id AS id,\n"
              "           alternate_geonames.name AS name,\n"
              "           length(alternate_geonames.name) AS l\n"
              "    FROM geonames, alternate_geonames, keywords_has_geonames\n"
              "    WHERE geonames.id=alternate_geonames.geonames_id\n"
              "    AND keywords_has_geonames.geonames_id=geonames.id\n"
              "    AND keywords_has_geonames.keyword=";
          sql += conn->quote(theKeyword);
          sql += "    AND alternate_geonames.language=";
          sql += conn->quote(theOptions.GetLanguage());
          sql +=
              "    ORDER BY preferred DESC,l\n"
              "  )\n"
              "  AS altname_tmp\n"
              "  GROUP BY id, name \n"
              ")\n"
              "AS altname_translations\n"
              "ON (georesults.id=altname_translations.id)\n"

              // -- left join to add alternate country name if available

              "LEFT JOIN\n"
              "(\n"
              "  SELECT iso2,name FROM\n"
              "  (\n"
              "    SELECT countries_iso2 AS iso2,\n"
              "           alternate_geonames.name AS name,\n"
              "           length(alternate_geonames.name) AS l\n"
              "    FROM geonames, alternate_geonames\n"
              "    WHERE geonames.features_code='PCLI'\n"
              "    AND geonames.id=alternate_geonames.geonames_id\n"
              "    AND alternate_geonames.language=";
          sql += conn->quote(theOptions.GetLanguage());
          sql +=
              "    ORDER BY preferred DESC,l\n"
              "  )\n"
              "  AS iso2_tmp\n"
              "  GROUP BY iso2, name\n"
              ")\n"
              "AS iso2_translations\n"
              "ON (georesults.iso2=iso2_translations.iso2)\n"
              "ORDER BY id;\n";
        }
        break;
      }
      case eCountKeywordLocations:
      {
        auto theKeyword = boost::any_cast<string>(theParams.at(eKeyword));

        sql += "SELECT count(*) AS count FROM keywords_has_geonames WHERE keyword=";
        sql += conn->quote(theKeyword);
        break;
      }
        // With -Weverything complains as all cases are covered
        //    default:
        //      break;
    }

    return sql;
  }
  catch (...)
  {
    throw Fmi::Exception::Trace(BCP, "Operation failed!");
  }
}

}  // namespace Locus

// ======================================================================

max_stars_repo_name: fmidev/smartmet-library-locus

max_issues_repo_name: fmidev/smartmet-library-locus

max_forks_repo_name: fmidev/smartmet-library-locus


text:
// Boost.Geometry (aka GGL, Generic Geometry Library)

// Copyright (c) 2007-2012 Barend Gehrels, Amsterdam, the Netherlands.
// Copyright (c) 2008-2012 Bruno Lalande, Paris, France.
// Copyright (c) 2009-2012 Mateusz Loskot, London, UK.

// Parts of Boost.Geometry are redesigned from Geodan's Geographic Library
// (geolib/GGL), copyright (c) 1995-2010 Geodan, Amsterdam, the Netherlands.

// Use, modification and distribution is subject to the Boost Software License,
// Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)

#ifndef BOOST_GEOMETRY_CORE_POINT_TYPE_HPP
#define BOOST_GEOMETRY_CORE_POINT_TYPE_HPP

#include <boost/mpl/assert.hpp>
#include <boost/range.hpp>
#include <boost/type_traits/remove_const.hpp>

#include <boost/geometry/core/ring_type.hpp>
#include <boost/geometry/core/tag.hpp>
#include <boost/geometry/core/tags.hpp>

namespace boost { namespace geometry
{

namespace traits
{

/*!
\brief Traits class indicating the type of contained points
\ingroup traits
\par Geometries:
    - all geometries except point
\par Specializations should provide:
    - typedef P type (where P should fulfil the Point concept)
\tparam Geometry geometry
*/
template <typename Geometry>
struct point_type
{
    BOOST_MPL_ASSERT_MSG
        (
            false, NOT_IMPLEMENTED_FOR_THIS_POINT_TYPE, (types<Geometry>)
        );
};

} // namespace traits

#ifndef DOXYGEN_NO_DISPATCH
namespace core_dispatch
{

template <typename Tag, typename Geometry>
struct point_type
{
    // Default: call traits to get point type
    typedef typename boost::remove_const
        <
            typename traits::point_type<Geometry>::type
        >::type type;
};

// Specialization for point: the point itself
template <typename Point>
struct point_type<point_tag, Point>
{
    typedef Point type;
};

// Specializations for linestring/ring, via boost::range
template <typename Linestring>
struct point_type<linestring_tag, Linestring>
{
    typedef typename boost::range_value<Linestring>::type type;
};

template <typename Ring>
struct point_type<ring_tag, Ring>
{
    typedef typename boost::range_value<Ring>::type type;
};

// Specialization for polygon: the point-type is the point-type of its rings
template <typename Polygon>
struct point_type<polygon_tag, Polygon>
{
    typedef typename point_type
        <
            ring_tag,
            typename ring_type<polygon_tag, Polygon>::type
        >::type type;
};

} // namespace core_dispatch
#endif // DOXYGEN_NO_DISPATCH

/*!
\brief \brief_meta{type, point_type, \meta_geometry_type}
\tparam Geometry \tparam_geometry
\ingroup core

\qbk{[include reference/core/point_type.qbk]}
*/
template <typename Geometry>
struct point_type
{
    typedef typename boost::remove_const<Geometry>::type ncg;
    typedef typename core_dispatch::point_type
        <
            typename tag<Geometry>::type,
            ncg
        >::type type;
};

}} // namespace boost::geometry

#endif // BOOST_GEOMETRY_CORE_POINT_TYPE_HPP

ext: hpp


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.

// PLASMA STORE: This is a simple object store server process
//
// It accepts incoming client connections on a unix domain socket
// (name passed in via the -s option of the executable) and uses a
// single thread to serve the clients. Each client establishes a
// connection and can create objects, wait for objects and seal
// objects through that connection.
//
// It keeps a hash table that maps object_ids (which are 20 byte long,
// just enough to store and SHA1 hash) to memory mapped files.

#include "ray/object_manager/plasma/store.h"

#include <limits.h>
#include <stdlib.h>
#include <string.h>

#include <boost/bind/bind.hpp>
#include <chrono>
#include <ctime>
#include <deque>
#include <memory>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

#include "absl/container/flat_hash_set.h"
#include "ray/common/asio/asio_util.h"
#include "ray/common/asio/instrumented_io_context.h"
#include "ray/object_manager/plasma/common.h"
#include "ray/object_manager/plasma/get_request_queue.h"
#include "ray/object_manager/plasma/malloc.h"
#include "ray/object_manager/plasma/plasma_allocator.h"
#include "ray/object_manager/plasma/protocol.h"
#include "ray/stats/metric_defs.h"
#include "ray/util/util.h"

namespace ph = boost::placeholders;
namespace fb = plasma::flatbuf;

namespace plasma {
namespace {

ray::ObjectID GetCreateRequestObjectId(const std::vector<uint8_t> &message) {
  uint8_t *input = (uint8_t *)message.data();
  size_t input_size = message.size();
  auto request = flatbuffers::GetRoot<fb::PlasmaCreateRequest>(input);
  RAY_DCHECK(plasma::VerifyFlatbuffer(request, input, input_size));
  return ray::ObjectID::FromBinary(request->object_id()->str());
}
}  // namespace

PlasmaStore::PlasmaStore(instrumented_io_context &main_service,
                         IAllocator &allocator,
                         const std::string &socket_name,
                         uint32_t delay_on_oom_ms,
                         float object_spilling_threshold,
                         ray::SpillObjectsCallback spill_objects_callback,
                         std::function<void()> object_store_full_callback,
                         ray::AddObjectCallback add_object_callback,
                         ray::DeleteObjectCallback delete_object_callback)
    : io_context_(main_service),
      socket_name_(socket_name),
      acceptor_(main_service, ParseUrlEndpoint(socket_name)),
      socket_(main_service),
      allocator_(allocator),
      add_object_callback_(add_object_callback),
      delete_object_callback_(delete_object_callback),
      object_lifecycle_mgr_(allocator_, delete_object_callback_),
      delay_on_oom_ms_(delay_on_oom_ms),
      object_spilling_threshold_(object_spilling_threshold),
      create_request_queue_(
          /*oom_grace_period_s=*/RayConfig::instance().oom_grace_period_s(),
          spill_objects_callback,
          object_store_full_callback,
          /*get_time=*/
          []() { return absl::GetCurrentTimeNanos(); },
          // absl can't check thread safety for lambda
          [this]() ABSL_NO_THREAD_SAFETY_ANALYSIS {
            mutex_.AssertHeld();
            return GetDebugDump();
          }),
      total_consumed_bytes_(0),
      get_request_queue_(
          io_context_,
          object_lifecycle_mgr_,
          // absl failed to check thread safety for lambda
          [this](const ObjectID &object_id, const auto &request)
              ABSL_NO_THREAD_SAFETY_ANALYSIS {
                mutex_.AssertHeld();
                this->AddToClientObjectIds(object_id, request->client);
              },
          [this](const auto &request) { this->ReturnFromGet(request); }) {
  const auto event_stats_print_interval_ms =
      RayConfig::instance().event_stats_print_interval_ms();
  if (event_stats_print_interval_ms > 0 && RayConfig::instance().event_stats()) {
    PrintAndRecordDebugDump();
  }
}

// TODO(pcm): Get rid of this destructor by using RAII to clean up data.
PlasmaStore::~PlasmaStore() {}

void PlasmaStore::Start() {
  // Start listening for clients.
  DoAccept();
}

void PlasmaStore::Stop() { acceptor_.close(); }

// If this client is not already using the object, add the client to the
// object's list of clients, otherwise do nothing.
void PlasmaStore::AddToClientObjectIds(const ObjectID &object_id,
                                       const std::shared_ptr<ClientInterface> &client) {
  // Check if this client is already using the object.
  auto &object_ids = client->GetObjectIDs();
  if (object_ids.find(object_id) != object_ids.end()) {
    return;
  }
  RAY_CHECK(object_lifecycle_mgr_.AddReference(object_id));
  // Add object id to the list of object ids that this client is using.
  client->MarkObjectAsUsed(object_id);
}

PlasmaError PlasmaStore::HandleCreateObjectRequest(const std::shared_ptr<Client> &client,
                                                   const std::vector<uint8_t> &message,
                                                   bool fallback_allocator,
                                                   PlasmaObject *object,
                                                   bool *spilling_required) {
  uint8_t *input = (uint8_t *)message.data();
  size_t input_size = message.size();
  ray::ObjectInfo object_info;
  fb::ObjectSource source;
  int device_num;
  ReadCreateRequest(input, input_size, &object_info, &source, &device_num);

  if (device_num != 0) {
    RAY_LOG(ERROR) << "device_num != 0 but CUDA not enabled";
    return PlasmaError::OutOfMemory;
  }

  auto error = CreateObject(object_info, source, client, fallback_allocator, object);
  if (error == PlasmaError::OutOfMemory) {
    RAY_LOG(DEBUG) << "Not enough memory to create the object " << object_info.object_id
                   << ", data_size=" << object_info.data_size
                   << ", metadata_size=" << object_info.metadata_size;
  }

  // Trigger object spilling if current usage is above the specified threshold.
  if (spilling_required != nullptr) {
    const int64_t footprint_limit = allocator_.GetFootprintLimit();
    if (footprint_limit != 0) {
      const float allocated_percentage =
          static_cast<float>(allocator_.Allocated()) / footprint_limit;
      if (allocated_percentage > object_spilling_threshold_) {
        RAY_LOG(DEBUG) << "Triggering object spilling because current usage "
                       << allocated_percentage << "% is above threshold "
                       << object_spilling_threshold_ << "%.";
        *spilling_required = true;
      }
    }
  }
  return error;
}

PlasmaError PlasmaStore::CreateObject(const ray::ObjectInfo &object_info,
                                      fb::ObjectSource source,
                                      const std::shared_ptr<Client> &client,
                                      bool fallback_allocator,
                                      PlasmaObject *result) {
  auto pair = object_lifecycle_mgr_.CreateObject(object_info, source, fallback_allocator);
  auto entry = pair.first;
  auto error = pair.second;
  if (entry == nullptr) {
    return error;
  }
  entry->ToPlasmaObject(result, /* check sealed */ false);
  // Record that this client is using this object.
  AddToClientObjectIds(object_info.object_id, client);
  return PlasmaError::OK;
}

void PlasmaStore::ReturnFromGet(const std::shared_ptr<GetRequest> &get_request) {
  // If the get request is already removed, do no-op. This can happen because the boost
  // timer is not atomic. See https://github.com/ray-project/ray/pull/15071.
  if (get_request->IsRemoved()) {
    return;
  }

  // Figure out how many file descriptors we need to send.
  absl::flat_hash_set<MEMFD_TYPE> fds_to_send;
  std::vector<MEMFD_TYPE> store_fds;
  std::vector<int64_t> mmap_sizes;
  for (const auto &object_id : get_request->object_ids) {
    const PlasmaObject &object = get_request->objects[object_id];
    MEMFD_TYPE fd = object.store_fd;
    if (object.data_size != -1 && fds_to_send.count(fd) == 0 && fd.first != INVALID_FD) {
      fds_to_send.insert(fd);
      store_fds.push_back(fd);
      mmap_sizes.push_back(object.mmap_size);
      if (get_request->is_from_worker) {
        total_consumed_bytes_ += object.data_size + object.metadata_size;
      }
    }
  }
  // Send the get reply to the client.
  Status s = SendGetReply(std::dynamic_pointer_cast<Client>(get_request->client),
                          &get_request->object_ids[0],
                          get_request->objects,
                          get_request->object_ids.size(),
                          store_fds,
                          mmap_sizes);
  // If we successfully sent the get reply message to the client, then also send
  // the file descriptors.
  if (s.ok()) {
    // Send all of the file descriptors for the present objects.
    for (MEMFD_TYPE store_fd : store_fds) {
      Status send_fd_status = get_request->client->SendFd(store_fd);
      if (!send_fd_status.ok()) {
        RAY_LOG(ERROR) << "Failed to send mmap results to client on fd "
                       << get_request->client;
      }
    }
  } else {
    RAY_LOG(ERROR) << "Failed to send Get reply to client on fd " << get_request->client;
  }
}

void PlasmaStore::ProcessGetRequest(const std::shared_ptr<Client> &client,
                                    const std::vector<ObjectID> &object_ids,
                                    int64_t timeout_ms,
                                    bool is_from_worker) {
  get_request_queue_.AddRequest(client, object_ids, timeout_ms, is_from_worker);
}

int PlasmaStore::RemoveFromClientObjectIds(const ObjectID &object_id,
                                           const std::shared_ptr<Client> &client) {
  auto &object_ids = client->GetObjectIDs();
  auto it = object_ids.find(object_id);
  if (it != object_ids.end()) {
    client->MarkObjectAsUnused(*it);
    RAY_LOG(DEBUG) << "Object " << object_id << " no longer in use by client";
    // Decrease reference count.
    object_lifecycle_mgr_.RemoveReference(object_id);
    // Return 1 to indicate that the client was removed.
    return 1;
  } else {
    // Return 0 to indicate that the client was not removed.
    return 0;
  }
}

void PlasmaStore::ReleaseObject(const ObjectID &object_id,
                                const std::shared_ptr<Client> &client) {
  auto entry = object_lifecycle_mgr_.GetObject(object_id);
  RAY_CHECK(entry != nullptr);
  // Remove the client from the object's array of clients.
  RAY_CHECK(RemoveFromClientObjectIds(object_id, client) == 1);
}

void PlasmaStore::SealObjects(const std::vector<ObjectID> &object_ids) {
  for (size_t i = 0; i < object_ids.size(); ++i) {
    RAY_LOG(DEBUG) << "sealing object " << object_ids[i];
    auto entry = object_lifecycle_mgr_.SealObject(object_ids[i]);
    RAY_CHECK(entry) << object_ids[i] << " is missing or not sealed.";
    add_object_callback_(entry->GetObjectInfo());
  }

  for (size_t i = 0; i < object_ids.size(); ++i) {
    get_request_queue_.MarkObjectSealed(object_ids[i]);
  }
}

int PlasmaStore::AbortObject(const ObjectID &object_id,
                             const std::shared_ptr<Client> &client) {
  auto &object_ids = client->GetObjectIDs();
  auto it = object_ids.find(object_id);
  if (it == object_ids.end()) {
    // If the client requesting the abort is not the creator, do not
    // perform the abort.
    return 0;
  }
  // The client requesting the abort is the creator. Free the object.
  RAY_CHECK(object_lifecycle_mgr_.AbortObject(object_id) == PlasmaError::OK);
  client->MarkObjectAsUnused(*it);
  return 1;
}

void PlasmaStore::ConnectClient(const boost::system::error_code &error) {
  if (!error) {
    // Accept a new local client and dispatch it to the node manager.
    auto new_connection = Client::Create(
        // NOLINTNEXTLINE : handler must be of boost::AcceptHandler type.
        boost::bind(&PlasmaStore::ProcessMessage, this, ph::_1, ph::_2, ph::_3),
        std::move(socket_));
  }
  // We're ready to accept another client.
  DoAccept();
}

void PlasmaStore::DisconnectClient(const std::shared_ptr<Client> &client) {
  client->Close();
  RAY_LOG(DEBUG) << "Disconnecting client on fd " << client;
  // Release all the objects that the client was using.
  absl::flat_hash_map<ObjectID, const LocalObject *> sealed_objects;
  auto &object_ids = client->GetObjectIDs();
  for (const auto &object_id : object_ids) {
    auto entry = object_lifecycle_mgr_.GetObject(object_id);
    if (entry == nullptr) {
      continue;
    }

    if (entry->Sealed()) {
      // Add sealed objects to a temporary list of object IDs. Do not perform
      // the remove here, since it potentially modifies the object_ids table.
      sealed_objects[object_id] = entry;
    } else {
      // Abort unsealed object.
      object_lifecycle_mgr_.AbortObject(object_id);
    }
  }

  /// Remove all of the client's GetRequests.
  get_request_queue_.RemoveGetRequestsForClient(client);

  for (const auto &[object_id, _] : sealed_objects) {
    RemoveFromClientObjectIds(object_id, client);
  }

  create_request_queue_.RemoveDisconnectedClientRequests(client);
}

Status PlasmaStore::ProcessMessage(const std::shared_ptr<Client> &client,
                                   fb::MessageType type,
                                   const std::vector<uint8_t> &message) {
  absl::MutexLock lock(&mutex_);
  // TODO(suquark): We should convert these interfaces to const later.
  uint8_t *input = (uint8_t *)message.data();
  size_t input_size = message.size();
  ObjectID object_id;

  // Process the different types of requests.
  switch (type) {
  case fb::MessageType::PlasmaCreateRequest: {
    const auto &object_id = GetCreateRequestObjectId(message);
    const auto &request = flatbuffers::GetRoot<fb::PlasmaCreateRequest>(input);
    const size_t object_size = request->data_size() + request->metadata_size();

    // absl failed analyze mutex safety for lambda
    auto handle_create =
        [this, client, message](
            bool fallback_allocator, PlasmaObject *result, bool *spilling_required)
            ABSL_NO_THREAD_SAFETY_ANALYSIS {
              mutex_.AssertHeld();
              return HandleCreateObjectRequest(
                  client, message, fallback_allocator, result, spilling_required);
            };

    if (request->try_immediately()) {
      RAY_LOG(DEBUG) << "Received request to create object " << object_id
                     << " immediately";
      auto result_error = create_request_queue_.TryRequestImmediately(
          object_id, client, handle_create, object_size);
      const auto &result = result_error.first;
      const auto &error = result_error.second;
      if (SendCreateReply(client, object_id, result, error).ok() &&
          error == PlasmaError::OK && result.device_num == 0) {
        static_cast<void>(client->SendFd(result.store_fd));
      }
    } else {
      auto req_id =
          create_request_queue_.AddRequest(object_id, client, handle_create, object_size);
      RAY_LOG(DEBUG) << "Received create request for object " << object_id
                     << " assigned request ID " << req_id << ", " << object_size
                     << " bytes";
      ProcessCreateRequests();
      ReplyToCreateClient(client, object_id, req_id);
    }
  } break;
  case fb::MessageType::PlasmaCreateRetryRequest: {
    auto request = flatbuffers::GetRoot<fb::PlasmaCreateRetryRequest>(input);
    RAY_DCHECK(plasma::VerifyFlatbuffer(request, input, input_size));
    const auto &object_id = ObjectID::FromBinary(request->object_id()->str());
    ReplyToCreateClient(client, object_id, request->request_id());
  } break;
  case fb::MessageType::PlasmaAbortRequest: {
    RAY_RETURN_NOT_OK(ReadAbortRequest(input, input_size, &object_id));
    RAY_CHECK(AbortObject(object_id, client) == 1) << "To abort an object, the only "
                                                      "client currently using it "
                                                      "must be the creator.";
    RAY_RETURN_NOT_OK(SendAbortReply(client, object_id));
  } break;
  case fb::MessageType::PlasmaGetRequest: {
    std::vector<ObjectID> object_ids_to_get;
    int64_t timeout_ms;
    bool is_from_worker;
    RAY_RETURN_NOT_OK(ReadGetRequest(
        input, input_size, object_ids_to_get, &timeout_ms, &is_from_worker));
    ProcessGetRequest(client, object_ids_to_get, timeout_ms, is_from_worker);
  } break;
  case fb::MessageType::PlasmaReleaseRequest: {
    RAY_RETURN_NOT_OK(ReadReleaseRequest(input, input_size, &object_id));
    ReleaseObject(object_id, client);
  } break;
  case fb::MessageType::PlasmaDeleteRequest: {
    std::vector<ObjectID> object_ids;
    std::vector<PlasmaError> error_codes;
    RAY_RETURN_NOT_OK(ReadDeleteRequest(input, input_size, &object_ids));
    error_codes.reserve(object_ids.size());
    for (auto &object_id : object_ids) {
      error_codes.push_back(object_lifecycle_mgr_.DeleteObject(object_id));
    }
    RAY_RETURN_NOT_OK(SendDeleteReply(client, object_ids, error_codes));
  } break;
  case fb::MessageType::PlasmaContainsRequest: {
    RAY_RETURN_NOT_OK(ReadContainsRequest(input, input_size, &object_id));
    if (object_lifecycle_mgr_.IsObjectSealed(object_id)) {
      RAY_RETURN_NOT_OK(SendContainsReply(client, object_id, 1));
    } else {
      RAY_RETURN_NOT_OK(SendContainsReply(client, object_id, 0));
    }
  } break;
  case fb::MessageType::PlasmaSealRequest: {
    RAY_RETURN_NOT_OK(ReadSealRequest(input, input_size, &object_id));
    SealObjects({object_id});
    RAY_RETURN_NOT_OK(SendSealReply(client, object_id, PlasmaError::OK));
  } break;
  case fb::MessageType::PlasmaEvictRequest: {
    // This code path should only be used for testing.
    int64_t num_bytes;
    RAY_RETURN_NOT_OK(ReadEvictRequest(input, input_size, &num_bytes));
    int64_t num_bytes_evicted = object_lifecycle_mgr_.RequireSpace(num_bytes);
    RAY_RETURN_NOT_OK(SendEvictReply(client, num_bytes_evicted));
  } break;
  case fb::MessageType::PlasmaConnectRequest: {
    RAY_RETURN_NOT_OK(SendConnectReply(client, allocator_.GetFootprintLimit()));
  } break;
  case fb::MessageType::PlasmaDisconnectClient:
    RAY_LOG(DEBUG) << "Disconnecting client on fd " << client;
    DisconnectClient(client);
    return Status::Disconnected("The Plasma Store client is disconnected.");
    break;
  case fb::MessageType::PlasmaGetDebugStringRequest: {
    RAY_RETURN_NOT_OK(SendGetDebugStringReply(
        client, object_lifecycle_mgr_.EvictionPolicyDebugString()));
  } break;
  default:
    // This code should be unreachable.
    RAY_CHECK(0);
  }
  return Status::OK();
}

void PlasmaStore::DoAccept() {
  acceptor_.async_accept(
      socket_,
      boost::bind(&PlasmaStore::ConnectClient, this, boost::asio::placeholders::error));
}

void PlasmaStore::ProcessCreateRequests() {
  // Only try to process requests if the timer is not set. If the timer is set,
  // that means that the first request is currently not serviceable because
  // there is not enough memory. In that case, we should wait for the timer to
  // expire before trying any requests again.
  if (create_timer_) {
    return;
  }

  auto status = create_request_queue_.ProcessRequests();
  uint32_t retry_after_ms = 0;
  if (!status.ok()) {
    retry_after_ms = delay_on_oom_ms_;

    if (!dumped_on_oom_) {
      RAY_LOG(INFO) << "Plasma store at capacity\n" << GetDebugDump();
      dumped_on_oom_ = true;
    }
  } else {
    dumped_on_oom_ = false;
  }

  if (retry_after_ms > 0) {
    // Try to process requests later, after space has been made.
    create_timer_ = execute_after(
        io_context_,
        [this]() {
          absl::MutexLock lock(&mutex_);
          create_timer_ = nullptr;
          ProcessCreateRequests();
        },
        retry_after_ms);
  }
}

void PlasmaStore::ReplyToCreateClient(const std::shared_ptr<Client> &client,
                                      const ObjectID &object_id,
                                      uint64_t req_id) {
  PlasmaObject result = {};
  PlasmaError error;
  bool finished = create_request_queue_.GetRequestResult(req_id, &result, &error);
  if (finished) {
    RAY_LOG(DEBUG) << "Finishing create object " << object_id << " request ID " << req_id;
    if (SendCreateReply(client, object_id, result, error).ok() &&
        error == PlasmaError::OK && result.device_num == 0) {
      static_cast<void>(client->SendFd(result.store_fd));
    }
  } else {
    static_cast<void>(SendUnfinishedCreateReply(client, object_id, req_id));
  }
}

int64_t PlasmaStore::GetConsumedBytes() { return total_consumed_bytes_; }

bool PlasmaStore::IsObjectSpillable(const ObjectID &object_id) {
  absl::MutexLock lock(&mutex_);
  auto entry = object_lifecycle_mgr_.GetObject(object_id);
  if (!entry) {
    // Object already evicted or deleted.
    return false;
  }
  return entry->Sealed() && entry->GetRefCount() == 1;
}

void PlasmaStore::PrintAndRecordDebugDump() const {
  absl::MutexLock lock(&mutex_);
  RecordMetrics();
  RAY_LOG(INFO) << GetDebugDump();
  stats_timer_ = execute_after(
      io_context_,
      [this]() { PrintAndRecordDebugDump(); },
      RayConfig::instance().event_stats_print_interval_ms());
}

void PlasmaStore::RecordMetrics() const {
  // TODO(sang): Add metrics.
  object_lifecycle_mgr_.RecordMetrics();
}

std::string PlasmaStore::GetDebugDump() const {
  std::stringstream buffer;
  buffer << "========== Plasma store: =================\n";
  buffer << "Current usage: " << (allocator_.Allocated() / 1e9) << " / "
         << (allocator_.GetFootprintLimit() / 1e9) << " GB\n";
  buffer << "- num bytes created total: "
         << object_lifecycle_mgr_.GetNumBytesCreatedTotal() << "\n";
  auto num_pending_requests = create_request_queue_.NumPendingRequests();
  auto num_pending_bytes = create_request_queue_.NumPendingBytes();
  buffer << num_pending_requests << " pending objects of total size "
         << num_pending_bytes / 1024 / 1024 << "MB\n";
  object_lifecycle_mgr_.GetDebugDump(buffer);
  return buffer.str();
}

}  // namespace plasma

ext: cc

max_stars_repo_name: mgelbart/ray

max_issues_repo_name: mgelbart/ray

max_forks_repo_name: mgelbart/ray


text:
#include <boost/algorithm/string.hpp>
#include <cstdio>

#include "HpsLogHelper.h"
#include "RequestDecoder.h"
#include "HpsRequestMsg.h"
#include "HpsCommonFunc.h"
#include "CJson.h"
#include "UrlCode.h"

static char* g_pszOsapUser = "osap_user";
static char* g_pszMerchantName = "merchant_name";
static char* g_pszClientId = "client_id";

#define DO_NOTHING 0

namespace sdo
{
    namespace hps
    {

        CRequestDecoder* CRequestDecoder::s_instance = NULL;

        CRequestDecoder* CRequestDecoder::GetInstance()
        {
            HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);
            if (NULL == s_instance)
            {
                    s_instance = new CRequestDecoder;
            }

            return s_instance;
        }

        CRequestDecoder::CRequestDecoder()
        {
            HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);
        }

        void CRequestDecoder::Decode(SDecodeResult &result, char *pBuff, int nLen, const std::string & strRemoteIp, unsigned int dwRemotePort)
        {
            HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);
            string strOriInput(pBuff, nLen);
            int nResult = result.reqMsg.Decode(pBuff, nLen);
            if(nResult != 0)
            {
                HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s, Message Decode error result[%d]\n",__FUNCTION__, nResult);
                result.errorCode = ERROR_DEOCDE_REQUEST_ERROR;
                result.isSuccessful = false;
                return;
            }

            result.headerInfo.isKeepAlive = result.reqMsg.IsKeepAlive();
            result.headerInfo.nHttpVersion = result.reqMsg.GetHttpVersion();

            size_t pos = GetJsonRpcBodyStartPos(result.reqMsg);
            std::string rpcGuid = getguid();
            if (string::npos != pos)
            {
                result.requestType = E_JsonRpc_Request;
                DecodeJsonRpcBody(result.reqMsg, pos, strRemoteIp, dwRemotePort, rpcGuid, result);
            }
            else
            {
                result.requestType = E_Normal_Request;
                DecodeBody(result.reqMsg, strRemoteIp, dwRemotePort, rpcGuid, result);
            }

            for (vector<SRequestInfo>::iterator iter = result.requests.begin();
                 iter!=result.requests.end(); ++iter)
            {
                iter->strOriInput = strOriInput;
            }
        }

        size_t CRequestDecoder::GetJsonRpcBodyStartPos(CHpsRequestMsg& reqMsg)
        {
            HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);
            size_t pos = reqMsg.GetBody().find_first_of('[');
            if (pos == string::npos)
            {
                return string::npos;
            }

            std::string body = URLDecoder::decode(reqMsg.GetBody().substr(0, pos).c_str());
            std::string bodycopy;
            size_t spacepos1 = body.find_first_not_of(' ');
            size_t spacepos2;

            while (spacepos1 != string::npos)
            {
                spacepos2 = body.find_first_of(' ', spacepos1);
                if (spacepos2 != string::npos)
                {
                    bodycopy += body.substr(spacepos1, spacepos2-spacepos1);
                }
                else
                {
                    bodycopy += body.substr(spacepos1);
                    break;
                }

                spacepos1 = body.find_first_not_of(' ', spacepos2);
            }

            if ( (!bodycopy.empty())
                    && (bodycopy != "data=") )
            {
                return string::npos;
            }

            return pos;
        }

        void CRequestDecoder::DecodeBody(CHpsRequestMsg &reqMsg, const std::string & strRemoteIp,
                                           unsigned int dwRemotePort, std::string& rpcGuid, SDecodeResult &result)
        {
           HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);

            SRequestInfo sReq;
            SetCommonRequestInfo(reqMsg, sReq, strRemoteIp, dwRemotePort, rpcGuid);
            SetRestRequestInfoWithKeyValue(sReq.strUriAttribute, sReq);
            sReq.headers = reqMsg.GetHeaders();
            sReq.strUserAgent = reqMsg.GetUserAgent();
            sReq.requestType = E_Normal_Request;
            result.requests.push_back(sReq);
            result.isSuccessful = true;
        }

        void CRequestDecoder::DecodeJsonRpcBody(CHpsRequestMsg &reqMsg, size_t startPos, const string & strRemoteIp,
                                                  unsigned int dwRemotePort, std::string& rpcGuid, SDecodeResult &result)
        {
            HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);
            std::string body = reqMsg.GetBody();
            if (!reqMsg.IsPost())
            {
                body = body.substr(startPos);
            }

            body = URLDecoder::decode(body.c_str());

            std::vector<CJsonDecoder> vecDecoder;
            int len = body.length();
            CJsonDecoder jsonDecoder(body.c_str(), len);

            if (-1 == jsonDecoder.GetValue(vecDecoder))
            {
                result.isSuccessful = false;
                result.errorCode = ERROR_JSONRPC_PARSE_ERROR;
                return;
            }

            std::map<string, int> sequenceNoMap;
            std::map<string, int>::iterator map_iter;

            for (std::vector<CJsonDecoder>::iterator iter = vecDecoder.begin();
                    iter != vecDecoder.end(); ++iter)
            {
                SRequestInfo sReq;
                SetCommonRequestInfo(reqMsg, sReq, strRemoteIp, dwRemotePort, rpcGuid);
                ParseOneRequest(*iter, sReq);
                sReq.headers = reqMsg.GetHeaders();
                sReq.strUserAgent = reqMsg.GetUserAgent();
                sReq.strContentType = reqMsg.GetContentType();
                //check for same identifier
                map_iter = sequenceNoMap.find(sReq.sequenceNo);
                if (map_iter != sequenceNoMap.end())
                {
                    result.isSuccessful = false;
                    result.errorCode = ERROR_JSONRPC_ERROR_SAMEIDENTIFIER;
                    return;
                }
                else
                {
                    sequenceNoMap[sReq.sequenceNo] = 1;
                }
                sReq.requestType = E_JsonRpc_Request;
                result.requests.push_back(sReq);
            }
            jsonDecoder.DestroyJsonDecoder();

            result.isSuccessful = true;
        }

        void CRequestDecoder::ParseOneRequest(CJsonDecoder& jsonCoder, SRequestInfo& sReq)
        {
            HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);
            std::string value, method, params;
            if ( (-1 == jsonCoder.GetValue("jsonrpc", value))
                || (value != "2.0") )
            {
                HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s, can't find field 'jsonrpc'\n", __FUNCTION__);
                sReq.isVaild = false;
                sReq.nCode = ERROR_JSONRPC_INVAILD_REQUEST;
                return;
            }

            if ( (-1 == jsonCoder.GetValue("id", value))
                || (value.empty()) )
            {
                HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s, can't find field 'id'\n", __FUNCTION__);
                sReq.isVaild = false;
                sReq.nCode = ERROR_JSONRPC_INVAILD_REQUEST;
                return;
            }

            sReq.sequenceNo = value;

            if ( (-1 == jsonCoder.GetValue("method", method))
                || (method.empty()) )
            {
                HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s, can't find field 'method'\n", __FUNCTION__);
                sReq.isVaild = false;
                sReq.sequenceNo = "";
                sReq.nCode = ERROR_JSONRPC_INVAILD_REQUEST;
                return;
            }

            sReq.strUriCommand = method;

            EJsonValueType valueType;
            if (-1 == jsonCoder.GetValueType("params", valueType))
            {
                HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s, can't find field 'params'\n", __FUNCTION__);
                sReq.isVaild = false;
                sReq.sequenceNo = "";
                sReq.nCode = ERROR_JSONRPC_INVAILD_REQUEST;
                return;
            }
            else
            {
                if (JSONTYPE_STRING == valueType)
                {
                    if (-1 == jsonCoder.GetValue("params", params))
                    {
                        HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s, get string 'params' error\n", __FUNCTION__);
                        sReq.isVaild = false;
                        sReq.nCode = ERROR_JSONRPC_INVAILD_REQUEST;
                        return;
                    }
                    sReq.strUriAttribute = params;
                    SetRestRequestInfoWithKeyValue(params, sReq);
                }
                else if (JSONTYPE_OBJECT == valueType)
                {
                    CJsonDecoder tempJsonDecoder;
                    if ( -1 == jsonCoder.GetValue("params", tempJsonDecoder) )
                    {
                        HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s, get object 'params' error\n", __FUNCTION__);
                        sReq.isVaild = false;
                        sReq.nCode = ERROR_JSONRPC_INVAILD_REQUEST;
                        return;
                    }
                    sReq.strUriAttribute = "";
                    json_object* jroot = json_tokener_parse(tempJsonDecoder.ToJsonString());
                    if (!is_error(jroot))
                    {
                        json_object_object_foreach(jroot, key, val)
                        {
                            if (!sReq.strUriAttribute.empty())
                            {
                               sReq.strUriAttribute += "&";
                            }
                            sReq.strUriAttribute += key;
                            sReq.strUriAttribute += "=";
                            sReq.strUriAttribute += json_object_get_string(val);
                        }
                        json_object_put(jroot);
                    }
                    HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s, params[%s]\n", __FUNCTION__, sReq.strUriAttribute.c_str());
                    SetRestRequestInfoWithJson(tempJsonDecoder, sReq);
                }
                else
                {
                    HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s, unsupported type of 'params'\n", __FUNCTION__);
                    sReq.isVaild = false;
                    sReq.sequenceNo = "";
                    sReq.nCode = ERROR_JSONRPC_INVAILD_PARAMS;
                    return;
                }
            }

            sReq.isVaild = true;
        }

        void CRequestDecoder::SetCommonRequestInfo(CHpsRequestMsg &reqMsg, SRequestInfo& sReq, const std::string & strRemoteIp,unsigned int dwRemotePort, std::string& rpcGuid)
        {
            HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);
            sReq.strUriCommand = reqMsg.GetCommand();
            sReq.strUriAttribute = reqMsg.GetBody();
            sReq.strCookie = reqMsg.GetCookie();
            sReq.strHost = reqMsg.GetHost();
            sReq.isKeepAlive = reqMsg.IsKeepAlive();
            sReq.m_nHttpVersion = reqMsg.GetHttpVersion();
            sReq.strContentType = reqMsg.GetContentType();
            if(reqMsg.IsXForwardedFor())
            {
                sReq.strRemoteIp = reqMsg.GetXForwardedFor();
                sReq.dwRemotePort = reqMsg.GetXFFPort();
            }
            else
            {
                sReq.strRemoteIp = strRemoteIp;
                sReq.dwRemotePort = dwRemotePort;
            }

            sReq.strGuid = getguid();
            sReq.strRpcGuid = rpcGuid;
        }

        void CRequestDecoder::SetRestRequestInfoWithKeyValue(string& strKeyValue, SRequestInfo& sReq)
        {
            HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);
            gettimeofday_a(&(sReq.tStart),0);

            sReq.strFnCallback = sdo::hps::CHpsRequestMsg::GetValueByKey(strKeyValue, "fn_callback");

            sReq.strUserName = sdo::hps::CHpsRequestMsg::GetValueByKey(strKeyValue, g_pszMerchantName);
            if (sReq.strUserName.empty())
            {
                sReq.strUserName = sdo::hps::CHpsRequestMsg::GetValueByKey(strKeyValue, g_pszClientId);
                if (sReq.strUserName.empty())
                {
                    sReq.strUserName = sdo::hps::CHpsRequestMsg::GetValueByKey(strKeyValue, g_pszOsapUser);
                }
            }

            string strTimestamp = sdo::hps::CHpsRequestMsg::GetValueByKey(strKeyValue, "timestamp");
            if (!strTimestamp.empty())
            {
                sReq.nTimestamp = sReq.tStart.tv_sec + sReq.tStart.tv_usec/1000000
                                    - atoi(strTimestamp.c_str());
            }
            sReq.strComment = sdo::hps::CHpsRequestMsg::GetValueByKey(strKeyValue, "comment");
            sReq.isVaild = true;
        }

        void CRequestDecoder::SetRestRequestInfoWithJson(CJsonDecoder& jsonCoder, SRequestInfo& sReq)
        {
            HP_XLOG(XLOG_DEBUG,"CRequestDecoder::%s\n", __FUNCTION__);
            gettimeofday_a(&(sReq.tStart),0);

            jsonCoder.GetValue("fn_callback", sReq.strFnCallback);
            jsonCoder.GetValue(g_pszMerchantName, sReq.strUserName);

            if (sReq.strUserName.empty())
            {
                jsonCoder.GetValue(g_pszClientId, sReq.strUserName);
                if (sReq.strUserName.empty())
                {
                    jsonCoder.GetValue(g_pszOsapUser, sReq.strUserName);
                }
            }

            string strTimestamp("");
            jsonCoder.GetValue("timestamp", strTimestamp);
            if (!strTimestamp.empty())
            {
                sReq.nTimestamp = sReq.tStart.tv_sec + sReq.tStart.tv_usec/1000000
                                    - atoi(strTimestamp.c_str());
            }

            jsonCoder.GetValue("comment", sReq.strComment);
            sReq.isVaild = true;
        }
    }

}

ext: cpp

max_stars_repo_name: shrewdlin/BPE

max_issues_repo_name: shrewdlin/BPE

max_forks_repo_name: shrewdlin/BPE


text:
// Copyright (c) 2016 by Sony Computer Entertainment Inc.
// This file is subject to the terms and conditions defined in file
// 'LICENSE.txt', which is part of this source code package.

#ifndef SCANLIB_PROGRESS_HPP
#define SCANLIB_PROGRESS_HPP

#include <atomic>
#include <chrono>
#include <functional>
#include <iosfwd>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
//#include <vector>
#include <boost/circular_buffer.hpp>

namespace details {
    // ****************
    // * progress_bar *
    // ****************
    class progress_bar {
    public:
        explicit progress_bar (unsigned width = 40);
        void set_value (float value);
        std::ostream & write (std::ostream & os) const;

    private:
        unsigned const width_;
        float value_ = 0.0;
    };

    std::ostream & operator << (std::ostream & os, progress_bar const & prog);

    // *************************
    // * hours_minutes_seconds *
    // *************************
    class hours_minutes_seconds {
    public:
        hours_minutes_seconds (float secs);

        std::string to_string () const;
        std::ostream & write (std::ostream & os) const;

        unsigned hours () const { return hours_; }
        unsigned minutes () const { return minutes_; }
        unsigned seconds () const { return seconds_; }

    private:
        unsigned hours_;
        unsigned minutes_;
        unsigned seconds_;

        static constexpr auto const minute_threshold_for_no_seconds = 2U;
        static constexpr auto const seconds_in_minute = 60U;
        static constexpr auto const minutes_in_hour = 60U;
        static constexpr auto const seconds_in_hour = 60U * 60U;
    };

    std::ostream & operator << (std::ostream & os, hours_minutes_seconds const & hms);

    // *****************************
    // * time_remaining_calculator *
    // *****************************
    class time_remaining_calculator {
    public:
        explicit time_remaining_calculator (std::size_t size);

        float update (float speed);

        float mean () const { return stats_.mean (); }
        std::size_t samples () const { return stats_.samples (); }
        float standard_deviation () const { return stats_.stddev (); }

        class stats {
        public:
            // Adds 'x' to the sample collection.
            void push (float x);
            // Removes 'x' from the sample collection.
            void pop (float x);

            std::size_t samples () const { return samples_; }
            float mean () const { return mean_; }
            float stddev () const;

        private:
            float variance () const;

            std::size_t samples_ = 0U;
            float mean_ = 0.0F;
            float s_ = 0.0F;
        };

    private:
        boost::circular_buffer<float> buffer_;
        stats stats_;
    };
}

// *********
// * timer *
// *********
class timer {
public:
    timer (std::function <void ()> proc, std::chrono::milliseconds interval);
    ~timer () noexcept;
    void clear ();

private:
    void run (std::function <void ()> proc, std::chrono::milliseconds interval) noexcept;

    std::atomic <bool> running_ {true};
    std::thread thread_;
};

typedef std::shared_ptr <timer>  timer_ptr;

timer_ptr setInterval (std::function <void ()> proc, std::chrono::milliseconds interval);
void clearInterval (timer_ptr timer) noexcept;

// ****************
// * updater_intf *
// ****************
class updater_intf {
public:
    virtual ~updater_intf () noexcept = 0;

    virtual updater_intf & completed (unsigned /*c*/) { return *this; }
    virtual updater_intf & completed_incr (unsigned /*by*/ = 1) { return *this; }

    virtual updater_intf & total (unsigned /*value*/) { return *this; }
    virtual updater_intf & total_incr (unsigned /*by*/ = 1) { return *this; }

    virtual updater_intf & run () { return *this; }
};

// ******************
// * silent_updater *
// ******************
class silent_updater final : public updater_intf {
public:
    explicit silent_updater (char const * /*message*/ = nullptr);
    ~silent_updater () noexcept;
};

// ***********
// * updater *
// ***********
class updater final : public updater_intf {
public:
    explicit updater (char const * message = nullptr, unsigned total = 0);
    ~updater () noexcept override;

    unsigned completed () const;
    updater & completed (unsigned c) override;
    updater & completed_incr (unsigned by = 1) override;

    unsigned total () const;
    updater & total (unsigned value) override;
    updater & total_incr (unsigned by = 1) override;

    updater & run () override;

private:
    std::ostream & time_left (float const average, std::ostream & os) const;
    bool have_reliable_estimate (float const average) const;

    void update ();

    mutable std::mutex mutex_;

    details::progress_bar bar_;
    // time remaining estimates will be based on the completion rate that we've achieved
    // over the last 30 seconds.
    details::time_remaining_calculator time_remaining_;
    unsigned completed_ = 0;
    unsigned total_ = 0;

    unsigned old_completed_ = 0;
    std::size_t prev_width_ = 0;

    timer_ptr timer_;
};
#endif // SCANLIB_PROGRESS_HPP
// eof progress.hpp

max_stars_repo_name: SNSystems/Program-repository

max_issues_repo_name: SNSystems/Program-repository

max_forks_repo_name: SNSystems/Program-repository


text:
//==================================================================================================
/*!

  Copyright 2016 NumScale SAS

  Distributed under the Boost Software License, Version 1.0.
  (See accompanying file LICENSE.md or copy at http://boost.org/LICENSE_1_0.txt)
*/
//==================================================================================================
#include <boost/simd/function/pack_cast.hpp>
#include <boost/simd/pack.hpp>
#include <simd_test.hpp>
#include <boost/simd/function/enumerate.hpp>

namespace bs = boost::simd;
using bs::pack;

template <typename T, typename Env>
void test_type(Env& runtime)
{
  static const std::size_t N = T::static_size;
  STF_EXPR_IS(bs::pack_cast<std::int8_t  >(T()), (bs::pack<std::int8_t,   N>));
  STF_EXPR_IS(bs::pack_cast<std::int16_t >(T()), (bs::pack<std::int16_t,  N>));
  STF_EXPR_IS(bs::pack_cast<std::int32_t >(T()), (bs::pack<std::int32_t,  N>));
  STF_EXPR_IS(bs::pack_cast<std::int64_t >(T()), (bs::pack<std::int64_t,  N>));
  STF_EXPR_IS(bs::pack_cast<std::uint8_t >(T()), (bs::pack<std::uint8_t,  N>));
  STF_EXPR_IS(bs::pack_cast<std::uint16_t>(T()), (bs::pack<std::uint16_t, N>));
  STF_EXPR_IS(bs::pack_cast<std::uint32_t>(T()), (bs::pack<std::uint32_t, N>));
  STF_EXPR_IS(bs::pack_cast<std::uint64_t>(T()), (bs::pack<std::uint64_t, N>));
  STF_EXPR_IS(bs::pack_cast<float        >(T()), (bs::pack<float        , N>));
  STF_EXPR_IS(bs::pack_cast<double       >(T()), (bs::pack<double       , N>));
}

STF_CASE_TPL( "Check cast return type", STF_NUMERIC_TYPES )
{
  static const std::size_t N = bs::pack<T>::static_size;

  test_type<bs::pack<T, N>>(runtime);
  test_type<bs::pack<T, N/2>>(runtime);
  test_type<bs::pack<T, N*2>>(runtime);
}