ReducerFunctionalObject.hpp

Go to the documentation of this file.

#ifndef REDUCERFUNCTIONALOBJECT_HPP_
#define REDUCERFUNCTIONALOBJECT_HPP

#include "ReducingHandlerBuilder.hpp"
#include "PocoJSONReducingInputMessageConvertor.hpp"
#include <Poco/SharedPtr.h>

namespace HCE{
namespace reduce{

const int OK = 0;
const int WRONG_JSON_STRUCT_ERROR = 1;
const int NOT_FOUND_ITEM_ID_ERROR = 2;
const std::string WRONG_JSON_STRUCT_DESCR = "err string";
const std::string NOT_FOUND_ITEM_ID_DESCR = "err string2";

class ReducerFunctionalObject{
  public:
    ReducerFunctionalObject();
    ~ReducerFunctionalObject();

    void accumulate(std::string& jsonReducingInputMessage, unsigned long long itemId);
    bool isCompleteTask(unsigned long long itemId, int nodesCount,
                        unsigned int ttl = USE_TASK_TTL_VALUE);
    Poco::SharedPtr<HCE::reduce_types::ReducingOutputMessage> reduce(unsigned long long itemId);

    unsigned int getTasksNumber();
    int getErrCode() const;
    std::string getErrMsg() const;

    int cleanupExpiredTasksByTTL(unsigned int maxRemoveTasks = 1,
                                 unsigned int ttl = USE_TASK_TTL_VALUE);
    unsigned int cleanupExpiredTasksByTTLQueue(unsigned int maxRemoveItems,
                                               unsigned int minTerminateTime );
    void getExceededTTLTasks( unsigned int maxTaskNumber,
                              std::vector<unsigned long long>& exceededTTLTasks,
                              unsigned int ttl = USE_TASK_TTL_VALUE);
    int getRejectedMessages() const;

  private:
    Poco::SharedPtr<HCE::reduce::ReducingHandler>_reducingHandler;
    HCE::reduce_convertors::PocoJSONReducingInputMessageConvertor _pocoJSONReducingInputMessageConvertor;
    int errCode;
    std::string errMsg;
    int _rejectedMessages;
};

}
}

#endif