SphinxFunctionalObject.cpp

Go to the documentation of this file.

#include <sstream>
#include <unistd.h>
#include <Poco/AutoPtr.h>
#include <Poco/String.h>
#include <Poco/Logger.h>

#include "SphinxFunctionalObject.hpp"
#include "SphinxError.hpp"
#include "SphinxConfigCreator.hpp"
#include "SphinxSchemaFile.hpp"
#include "SphinxInputJsonMessage.hpp"
#include "SphinxOutputJsonMessage.hpp"
#include "SphinxMessageConst.hpp"
#include "SphinxAdminCommand.hpp"
#include "SphinxInputJsonMessageSearch.hpp"

namespace HCE
{
namespace sphinx
{
//-----------------------------------------------------------------------------
SphinxFunctionalObject::SphinxFunctionalObject(const std::string& nodeName_, const std::string& homeDir_, const std::string& indexName_, bool startSearchd_, bool stopSearchd_)
:inherited(), SphinxNodeOptions(nodeName_, 0, homeDir_, indexName_), searcher(*this, startSearchd_, stopSearchd_), indexer(*this),
 loaded(false), checkIncomingDocuments(false), packIncomingDocuments(false), minNumberFieldsPacking(0),
 logger(sphinx_search_const::moduleName), messagesCollection(message_const::messages), message(messagesCollection),
 matchMode(SPH_MATCH_EXTENDED2), sortMode(SPH_SORT_RELEVANCE), rankingMode(SPH_RANK_DEFAULT), rankerExpression("")
{
  if (!homeDir_.empty() && !indexName_.empty())
    if (loadConfiguration(indexName_))
    {
      log(LoggerStream::Priority::PRIO_DEBUG) << message(message_const::LOAD_CONFIG_SUCCESS)
          << message(message_const::HOST, getServerHost()) << message(message_const::PORT, getServerPort()) << flush;
      if (startSearchd_)
        startSearchd();
      else
      {
        _isError = !easyStart();
      }
    }
}
//-----------------------------------------------------------------------------
SphinxFunctionalObject::SphinxFunctionalObject(const std::string& nodeName_, unsigned int nodeNumber_, const std::string& homeDir_, const std::string& indexName_, bool startSearchd_, bool stopSearchd_)
:inherited(), SphinxNodeOptions(nodeName_, nodeNumber_, homeDir_, indexName_), searcher(*this, startSearchd_, stopSearchd_), indexer(*this),
 loaded(false), checkIncomingDocuments(false), packIncomingDocuments(false), minNumberFieldsPacking(0),
 logger(sphinx_search_const::moduleName), messagesCollection(message_const::messages), message(messagesCollection),
 matchMode(SPH_MATCH_EXTENDED2), sortMode(SPH_SORT_RELEVANCE), rankingMode(SPH_RANK_DEFAULT), rankerExpression("")
{
  if (!homeDir_.empty() && !indexName_.empty())
    if (loadConfiguration(indexName_))
    {
      log(LoggerStream::Priority::PRIO_DEBUG) << message(message_const::LOAD_CONFIG_SUCCESS) << message(message_const::INDEX, getIndexName())
          << message(message_const::HOST, getServerHost()) << message(message_const::PORT, getServerPort()) << flush;
      if (startSearchd_)
        startSearchd();
      else
      {
        _isError = !easyStart();
      }
    }
}
//-----------------------------------------------------------------------------
SphinxFunctionalObject::~SphinxFunctionalObject(void)
{
  if (searcher.isNeedStopSearchd())
//    if (searcher.isActiveSearchd())
      searcher.stopSearchd();
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::resetLogStream(void)
{
  logger.resetLogStream();
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::setLogStream(std::ostream& os)
{
  logger.setLogStream(os);
}
//-----------------------------------------------------------------------------
std::string SphinxFunctionalObject::logMsg(bool isReset)
{
  return logger.logMsg(isReset);
}
//-----------------------------------------------------------------------------
std::ostream& SphinxFunctionalObject::log(LoggerStream::Priority logPriority)
{
  return logger.log(logPriority);
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::setLoggable(Loggable loggable)
{
  logger.setLoggable(loggable);
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::resetLoggable(void)
{
  logger.resetLoggable();
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::parseHostPort(const std::string& in, std::string& outHost, unsigned int& outPort) throw (std::exception)
{
  size_t pos = in.find(':');
  if (pos==std::string::npos)
  {
    outHost = sphinx_search_const::defaultHost;
    outPort = std::stoul(Poco::trim(in));
  }
  else
  {
    outHost = Poco::trim(std::string(in, 0, pos));
    outPort = std::stoul(Poco::trim(std::string(in, pos+1)));
  }
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::setRankingMode(unsigned int rankingMode_, const std::string& rankerExpression_)
{
  rankingMode=rankingMode_;
  if (!rankerExpression_.empty())
    rankerExpression = rankerExpression_;
}
//-----------------------------------------------------------------------------
bool SphinxFunctionalObject::loadConfiguration(const std::string& indexName)
{
  loaded = false;
  _isError = false;
  errorMsg.clear();
  errorCode = NO_ERROR;
  std::string configFile = getDataDir()+"/"+indexName+"/"+sphinx_search_const::sphinxPropertyFile;

  try
  {
    Poco::AutoPtr<SphinxConfigCreator> pConf = new SphinxConfigCreator(configFile);

    rankingMode = pConf->getUInt(sphinx_admin_command_const::optionsRanker, SPH_RANK_DEFAULT);
    rankerExpression = pConf->getString(sphinx_admin_command_const::optionsRankerExpression);

    const std::string listen = pConf->getString(sphinx_search_const::searchdListen);
    const std::string maxMatches = pConf->getString(sphinx_search_const::searchdMaxMatches);

    std::string host;
    unsigned int port = 0;

    parseHostPort(listen, host, port);

    setServerHost((host.empty())?sphinx_search_const::defaultHost:host);
    setServerPort((port==0)?sphinx_search_const::defaultPort:port);

    std::string file = getDataDir()+"/"+indexName+"/"+sphinx_search_const::sphinxSchemaFile;
    std::ifstream ifs(file.c_str());
    if (!ifs.is_open())
      throw Poco::Exception(message(message_const::FILE_NOT_FOUND, file), ERROR_LOAD_CONFIG);

    std::vector<std::pair<std::string, int> > weights;
    SphinxSchemaFile schemaFile(ifs);
    loaded = schemaFile.getFields(weights);
    ifs.close();
    if (!loaded)
      throw Poco::Exception(schemaFile.getErrorMsg(), ERROR_LOAD_CONFIG);

    ifs.open(file.c_str());
    if (!ifs.is_open())
      throw Poco::Exception(message(message_const::FILE_NOT_OPEN, file), ERROR_LOAD_CONFIG);

    SphinxSchemaFile schema(ifs);
    std::vector<std::string> schemaNames;
    loaded = schema.getNames(schemaNames);
    ifs.close();
    if (!loaded)
      throw Poco::Exception(schemaFile.getErrorMsg(), ERROR_LOAD_CONFIG);

    const unsigned int attributesCount = schema.getAttributesCount();
    log(LoggerStream::Priority::PRIO_DEBUG) << message(message_const::SEARCHD_MAX_MATCHES, maxMatches) << flush;
    log(LoggerStream::Priority::PRIO_DEBUG) << message(message_const::SCHEMA_NAMES_SIZE, (unsigned int)schemaNames.size()) << flush;
    log(LoggerStream::Priority::PRIO_DEBUG) << message(message_const::ATTRIBUTES_COUNT, attributesCount) << flush;

    // realloc memory for manager
    searcher.resizeResultData(input_json_message_const::defaultMaxResultsNumber, attributesCount + input_json_message_const::extraAttributesNumber); // +2 need for ID docs and weight

    searcher.setFullSchemaNames(schemaNames);
    loaded = false;
    if (searcher.setFieldWeights(weights))
    {
      weights.clear();

      std::vector<std::string> keys;
      pConf->keys(keys);

      for (size_t i = 0;i<keys.size();++i)
      {
        size_t found = keys[i].find(sphinx_search_const::indexNameField);
        if (found!=std::string::npos)
        {
          found = keys[i].find_first_not_of(' ', found+sphinx_search_const::indexNameField.length());
          if (found!=std::string::npos)
          {
            weights.push_back(std::make_pair(keys[i].substr(found), 1));
          }
        }
      }
      if (weights.empty())
        throw Poco::Exception(message(message_const::EMPTY_LIST_OF_INDEX_FILES), ERROR_LOAD_CONFIG);

      loaded = searcher.setIndexWeights(weights);
    }
  }
  catch(Poco::Exception& e)
  {
    applyError(e.message(), e.code());
    log(LoggerStream::Priority::PRIO_DEBUG) << message(message_const::LOAD_CONFIG_FAIL, errorMsg) << flush;
  }
  catch(std::exception& e)
  {
    applyError(e.what(), ERROR_LOAD_CONFIG);
    log(LoggerStream::Priority::PRIO_DEBUG) << message(message_const::LOAD_CONFIG_FAIL, errorMsg) << flush;
  }
  return loaded;
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::resizeResultData(size_t matchesCount, size_t attributesCount)
{
  searcher.resizeResultData(matchesCount, attributesCount);
}
//-----------------------------------------------------------------------------
bool SphinxFunctionalObject::easyStart(void)
{
  return searcher.easyStart(matchMode, sortMode, rankingMode, rankerExpression);
}
//-----------------------------------------------------------------------------
bool SphinxFunctionalObject::startSearchd(void)
{
  return searcher.startSearchd();
}
//-----------------------------------------------------------------------------
bool SphinxFunctionalObject::stopSearchd(void)
{
  return searcher.stopSearchd();
}
//-----------------------------------------------------------------------------
bool SphinxFunctionalObject::isActiveSearchd(void)
{
  return searcher.isActiveSearchd();
}
//-----------------------------------------------------------------------------
bool SphinxFunctionalObject::isConnected(void) const
{
  return searcher.isConnected();
}
//-----------------------------------------------------------------------------
bool SphinxFunctionalObject::open(void)
{
  searcher.setMatchMode(matchMode);
  searcher.setRankingMode(rankingMode, rankerExpression);
  searcher.setSortMode(sortMode);
  return searcher.open();
}
//-----------------------------------------------------------------------------
bool SphinxFunctionalObject::close(void)
{
  return searcher.close();
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::resetError(void)
{
  inherited::isError(false);
  inherited::setErrorMsg("");
  inherited::setErrorCode(NO_ERROR);
  searcher.isError(false);
  searcher.setErrorMsg("");
  searcher.setErrorCode(NO_ERROR);
  indexer.isError(false);
  indexer.setErrorMsg("");
  indexer.setErrorCode(NO_ERROR);
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::applyError(const std::string& msg, unsigned int code)
{
  _isError = true;
  inherited::setErrorMsg(msg);
  inherited::setErrorCode(code);
}
//-----------------------------------------------------------------------------
std::string SphinxFunctionalObject::getErrorMsg(void) const
{
  std::string error;
  if (inherited::isError())
    error = inherited::getErrorMsg();
  else if (searcher.isError())
    error = searcher.getErrorMsg();
  else if (indexer.isError())
    error = indexer.getErrorMsg();
  return error;
}
//-----------------------------------------------------------------------------
unsigned int SphinxFunctionalObject::getErrorCode(void) const
{
  unsigned int code = NO_ERROR;
  if (inherited::isError())
    code = inherited::getErrorCode();
  else if (searcher.isError())
    code = searcher.getErrorCode();
  else if (indexer.isError())
    code = indexer.getErrorCode();
  return code;
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::setServerHost(const std::string& serverHost_)
{
  searcher.setServerHost(serverHost_);
}
//-----------------------------------------------------------------------------
std::string SphinxFunctionalObject::getServerHost(void) const
{
  return searcher.getServerHost();
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::setServerPort(int serverPort_)
{
  searcher.setServerPort(serverPort_);
}
//-----------------------------------------------------------------------------
int SphinxFunctionalObject::getServerPort(void) const
{
  return searcher.getServerPort();
}
//-----------------------------------------------------------------------------
bool SphinxFunctionalObject::setIndexFileName(const std::string& indexFileName_)
{
  return searcher.setIndexFileName(indexFileName_);
}
//-----------------------------------------------------------------------------
std::string SphinxFunctionalObject::getIndexFileName(void) const
{
  return searcher.getIndexFileName();
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::setTimeoutedRequests(unsigned long long timeoutedRequests_)
{
  searcher.setTimeoutedRequests(timeoutedRequests_);
}
//-----------------------------------------------------------------------------
unsigned long long SphinxFunctionalObject::getTimeoutedRequests(void) const
{
  return searcher.getTimeoutedRequests();
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::resetTimeoutedRequests(void)
{
  searcher.resetTimeoutedRequests();
}
//-----------------------------------------------------------------------------
void SphinxFunctionalObject::setDefaultRequestTimeout(unsigned long defaultRequestTimeout_)
{
  searcher.setDefaultRequestTimeout(defaultRequestTimeout_);
}
//-----------------------------------------------------------------------------
unsigned long SphinxFunctionalObject::getDefaultRequestTimeout(void) const
{
  return searcher.getDefaultRequestTimeout();
}
//-----------------------------------------------------------------------------
std::string SphinxFunctionalObject::Process(const std::string& json)
{
  _isError = false;
  errorMsg = "";
  errorCode = NO_ERROR;
  std::string resultData;
  timeval tvStart, tvStop;
  gettimeofday(&tvStart, 0); //gettimeofday does not support TZ adjust on Linux.
  bool needDefaultJson = false;
  try
  {
    SphinxInputJsonMessage inputJsonMessage(json);
    if (inputJsonMessage.isError())
      throw Poco::Exception(inputJsonMessage.getErrorMsg(), PARSE_ERROR);

    if (inputJsonMessage.getType() == SphinxInputJsonMessage::MessageType::mtSearch)
    {
      needDefaultJson = true;
      resultData = searcher.makeSearchCommand(inputJsonMessage.getData());
      if (resultData.empty())
        resultData = makeDefaultJSON(message(message_const::EMPTY_RESULT_DATA));
    }
    else if (inputJsonMessage.getType() == SphinxInputJsonMessage::MessageType::mtIndex)
    {
      indexer.makeIndexCommand(inputJsonMessage.getData());
    }
    else if (inputJsonMessage.getType() == SphinxInputJsonMessage::MessageType::mtManage)
    {
      indexer.makeAdminCommand(inputJsonMessage.getData(), resultData);
    }
    else
      throw Poco::Exception(message(message_const::INCORRECT_JSON_TYPE), COMMAND_ERROR);

  }
  catch(Poco::Exception& e)
  {
    applyError(e.message(), e.code());
    if (needDefaultJson)
      resultData = makeDefaultJSON(message(message_const::ERROR, getErrorMsg()));
    Poco::Logger::root().log(e);
  }
  catch(std::exception& e)
  {
    applyError(e.what(), COMMAND_ERROR);
    if (needDefaultJson)
      resultData = makeDefaultJSON(message(message_const::ERROR, getErrorMsg()));
    Poco::Logger::root().error(e.what());
  }

  std::string resJson;
  gettimeofday(&tvStop, 0);

  SphinxOutputJsonMessage outputMessage;
  outputMessage.setErrorCode(getErrorCode());
  outputMessage.setErrorMessage(getErrorMsg());
  outputMessage.setData(resultData);
  outputMessage.setTime(SphinxSearcher::getTimeInterval(tvStart, tvStop)/1000);
  outputMessage.serialize(resJson);

  if (outputMessage.isError())
    log(LoggerStream::Priority::PRIO_DEBUG) << message(message_const::SERIALIZE_OUTPUT_JSON_MESSAGE)
    << message(message_const::ERROR_MSG, outputMessage.getErrorMsg())
    << message(message_const::ERROR_CODE, outputMessage.getErrorCode()) << flush;

  return resJson;
}
//-----------------------------------------------------------------------------
SphinxOutputJsonMessage SphinxFunctionalObject::Process(SphinxInputJsonMessage& inputJsonMessage)
{
  _isError = false;
  errorMsg.clear();
  errorCode = NO_ERROR;
  std::string resultData;
  timeval tvStart, tvStop;
  gettimeofday(&tvStart, 0);          //gettimeofday does not support TZ adjust on Linux.
  bool needDefaultJson = false;
  try
  {
    if (inputJsonMessage.getType() == SphinxInputJsonMessage::MessageType::mtSearch)
    {
      needDefaultJson = true;
      resultData = searcher.makeSearchCommand(inputJsonMessage.getData());
      if (resultData.empty())
        resultData = makeDefaultJSON(message(message_const::EMPTY_RESULT_DATA));
    }
    else if (inputJsonMessage.getType() == SphinxInputJsonMessage::MessageType::mtIndex)
    {
      indexer.makeIndexCommand(inputJsonMessage.getData());
    }
    else if (inputJsonMessage.getType() == SphinxInputJsonMessage::MessageType::mtManage)
    {
      indexer.makeAdminCommand(inputJsonMessage.getData(), resultData);
    }
    else
      throw Poco::Exception(message(message_const::INCORRECT_JSON_TYPE), COMMAND_ERROR);
  }
  catch(Poco::Exception& e)
  {
    applyError(e.message(), e.code());
    if (needDefaultJson)
      resultData = makeDefaultJSON(message(message_const::ERROR, getErrorMsg()));
    Poco::Logger::root().log(e);
  }
  catch(std::exception& e)
  {
    applyError(e.what(), COMMAND_ERROR);
    if (needDefaultJson)
      resultData = makeDefaultJSON(message(message_const::ERROR, getErrorMsg()));
    Poco::Logger::root().error(e.what());
  }

  gettimeofday(&tvStop, 0);

  SphinxOutputJsonMessage outputMessage;
  outputMessage.setErrorCode(getErrorCode());
  outputMessage.setErrorMessage(getErrorMsg());
  outputMessage.setData(resultData);
  outputMessage.setTime(SphinxSearcher::getTimeInterval(tvStart, tvStop)/1000);

  return outputMessage;
}
//-----------------------------------------------------------------------------
std::string SphinxFunctionalObject::makeDefaultJSON(const std::string& logMsg)
{
  log(LoggerStream::Priority::PRIO_DEBUG) << logMsg << " " << message(message_const::GENERATE_DEFAULT_JSON) << flush;
  SphinxDefaultJSON defaultJson;
  return defaultJson.getJSON();
}
//-----------------------------------------------------------------------------
std::string SphinxFunctionalObject::makeErrorMessage(const std::string& inputMessage, unsigned int errorCode, const std::string& errorMessage)
{
  std::string resultMessageBody;
  HCE::sphinx::SphinxOutputJsonMessage outputJsonMessage;
  HCE::sphinx::SphinxInputJsonMessage inputJsonMessage(inputMessage);
  if ((inputJsonMessage.getType() == HCE::sphinx::SphinxInputJsonMessage::MessageType::mtSearch) && !inputJsonMessage.isError())
  {
    unsigned int queryId = 0;
    HCE::sphinx::SphinxInputJsonMessageSearch messageSearch(inputJsonMessage.getData());
    for (size_t k=0;k<messageSearch.getQueryParameters().size();++k)
    {
      if (HCE::sphinx::SphinxSearchInterface::insensitiveCompare(messageSearch.getQueryParameters()[k].first, HCE::sphinx::input_json_message_const::queryId))
      {
        std::istringstream(messageSearch.getQueryParameters()[k].second) >> queryId;
        break;
      }
    }
    if (queryId)
    {
      HCE::sphinx::SphinxRequestInfo requestInfo;
      requestInfo.setQueryId(queryId);
      HCE::sphinx::SphinxResultData resultData;
      resultData.addRequestInfo(std::forward<HCE::sphinx::SphinxRequestInfo>(requestInfo));

      HCE::sphinx::SphinxResultDataSerializator resultSerializator(resultData);
      if (resultSerializator.serialize(resultMessageBody))
        outputJsonMessage.setData(resultMessageBody);
    }
  }

  outputJsonMessage.setErrorCode(errorCode);
  outputJsonMessage.setErrorMessage(errorMessage);

  if (!outputJsonMessage.serialize(resultMessageBody))
  {
    std::stringstream errMsg;
    errMsg << message_const::serializationOutputJsonMessageError << ": " << outputJsonMessage.getErrorMsg();
    Poco::Logger::root().log(Poco::Message(sphinx_search_const::moduleName, errMsg.str(), Poco::Message::Priority::PRIO_ERROR));
  }
  return resultMessageBody;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
} // namespace sphinx
} // namespace HCE